GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Framed encoding serde POC

Browse Source
This commit is contained in:
durch
2023-04-18 18:18:54 +02:00
parent f086f9c35a
commit d5e77e499b
3 changed files with 127 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/nymsphinx/framing/src/codec.rs
+107 -55
View File
@@ -27,6 +27,9 @@ pub enum SphinxCodecError {
#[error("could not convert to bytes")]
ToBytes,
#[error("could not convert to bytes")]
FromBytes,
}
#[derive(Error, Debug)]
@@ -70,6 +73,8 @@ impl Encoder<FramedNymPacket> for SphinxCodec {
type Error = SphinxCodecError;
fn encode(&mut self, item: FramedNymPacket, dst: &mut BytesMut) -> Result<(), Self::Error> {
let encoded_size = bincode::serialized_size(&item).unwrap() as u32;
dst.put_u32(encoded_size);
dst.put(
bincode::serialize(&item)
.map_err(|_| SphinxCodecError::ToBytes)?
@@ -98,33 +103,58 @@ impl Decoder for SphinxCodec {
return Ok(None);
}
// because header is so small and simple it makes no point in trying to cache
// this result. It will be just simpler to re-decode it
let header = match Header::decode(src)? {
Some(header) => header,
None => return Ok(None), // we have some data but not enough to get header back
let next_size = if let Some(size) = src.get(..4) {
u32::from_be_bytes(size.try_into().unwrap()) as usize
} else {
0
};
let sphinx_packet_size = header.packet_size.size();
let frame_len = header.size() + sphinx_packet_size;
if src.len() < frame_len {
// we don't have enough bytes to read the rest of frame
src.reserve(sphinx_packet_size);
// let next_size = src.get_u64() as usize;
if next_size == 0 {
return Ok(None);
}
if let Some(next_packet) = src.get(4..next_size + 4) {
match bincode::deserialize::<Self::Item>(next_packet) {
Ok(packet) => {
src.advance(next_size + 4);
return Ok(Some(packet));
}
Err(_) => {
println!("Could not decode packet");
return Ok(None);
}
}
}
Ok(None)
// because header is so small and simple it makes no point in trying to cache
// this result. It will be just simpler to re-decode it
// let header = match Header::decode(src)? {
// Some(header) => header,
// None => return Ok(None), // we have some data but not enough to get header back
// };
// let sphinx_packet_size = framed_packet.header.packet_size.size();
// let frame_len = framed_packet.header.size() + sphinx_packet_size;
// if src.len() < frame_len {
// // we don't have enough bytes to read the rest of frame
// src.reserve(sphinx_packet_size);
// return Ok(None);
// }
// advance buffer past the header - at this point we have enough bytes
src.advance(header.size());
let sphinx_packet_bytes = src.split_to(sphinx_packet_size);
// src.advance(header.size());
// let sphinx_packet_bytes = src.split_to(sphinx_packet_size);
// here it could be debatable whether stream is corrupt or not,
// but let's go with the safer approach and assume it is.
let packet = SphinxPacket::from_bytes(&sphinx_packet_bytes)?;
let nymsphinx_packet = FramedNymPacket {
header,
packet: NymPacket::Sphinx(packet),
};
// let packet = SphinxPacket::from_bytes(&sphinx_packet_bytes)?;
// let nymsphinx_packet = FramedNymPacket {
// header,
// packet: NymPacket::Sphinx(packet),
// };
// As per docs:
// Before returning from the function, implementations should ensure that the buffer
@@ -136,24 +166,24 @@ impl Decoder for SphinxCodec {
// reserve for that.
// we also assume the next packet coming from the same client will use exactly the same versioning
// as the current packet
let mut allocate_for_next_packet = header.size() + PacketSize::AckPacket.size();
if !src.is_empty() {
match Header::decode(src) {
Ok(Some(next_header)) => {
allocate_for_next_packet = next_header.size() + next_header.packet_size.size();
}
Ok(None) => {
// we don't have enough information to know how much to reserve, fallback to the ack case
}
// let mut allocate_for_next_packet = framed_packet.header.size() + PacketSize::AckPacket.size();
// if !src.is_empty() {
// match Header::decode(src) {
// Ok(Some(next_header)) => {
// allocate_for_next_packet = next_header.size() + next_header.packet_size.size();
// }
// Ok(None) => {
// // we don't have enough information to know how much to reserve, fallback to the ack case
// }
// the next frame will be malformed but let's leave handling the error to the next
// call to 'decode', as presumably, the current sphinx packet is still valid
Err(_) => return Ok(Some(nymsphinx_packet)),
};
}
src.reserve(allocate_for_next_packet);
// // the next frame will be malformed but let's leave handling the error to the next
// // call to 'decode', as presumably, the current sphinx packet is still valid
// Err(_) => return Ok(Some(framed_packet)),
// };
// }
// src.reserve(allocate_for_next_packet);
Ok(Some(nymsphinx_packet))
// Ok(Some(framed_packet))
}
}
@@ -207,7 +237,7 @@ mod packet_encoding {
let packet = FramedNymPacket {
header,
packet: NymPacket::Sphinx(sphinx_packet),
packet: Some(NymPacket::Sphinx(sphinx_packet)),
};
let mut bytes = BytesMut::new();
@@ -215,7 +245,7 @@ mod packet_encoding {
let decoded = SphinxCodec.decode(&mut bytes).unwrap().unwrap();
assert_eq!(decoded.header, header);
assert_eq!(decoded.packet.to_bytes().unwrap(), sphinx_bytes)
assert_eq!(decoded.packet.unwrap().to_bytes().unwrap(), sphinx_bytes)
}
#[cfg(test)]
@@ -295,16 +325,18 @@ mod packet_encoding {
packet_size: Default::default(),
packet_mode: Default::default(),
},
packet: NymPacket::Sphinx(make_valid_sphinx_packet(Default::default())),
packet: Some(NymPacket::Sphinx(make_valid_sphinx_packet(
Default::default(),
))),
};
let mut bytes = BytesMut::new();
SphinxCodec.encode(packet, &mut bytes).unwrap();
assert!(SphinxCodec.decode(&mut bytes).unwrap().is_some());
assert_eq!(
bytes.capacity(),
Header::LEGACY_SIZE + PacketSize::AckPacket.size()
);
// assert_eq!(
// bytes.capacity(),
// Header::LEGACY_SIZE + PacketSize::AckPacket.size()
// );
}
#[test]
@@ -312,16 +344,18 @@ mod packet_encoding {
// if full frame is used exactly, there should be enough space for header + ack packet
let packet = FramedNymPacket {
header: Header::default(),
packet: NymPacket::Sphinx(make_valid_sphinx_packet(Default::default())),
packet: Some(NymPacket::Sphinx(make_valid_sphinx_packet(
Default::default(),
))),
};
let mut bytes = BytesMut::new();
SphinxCodec.encode(packet, &mut bytes).unwrap();
assert!(SphinxCodec.decode(&mut bytes).unwrap().is_some());
assert_eq!(
bytes.capacity(),
Header::VERSIONED_SIZE + PacketSize::AckPacket.size()
);
// assert_eq!(
// bytes.capacity(),
// Header::VERSIONED_SIZE + PacketSize::AckPacket.size()
// );
}
#[test]
@@ -342,7 +376,9 @@ mod packet_encoding {
packet_size: Default::default(),
packet_mode: Default::default(),
},
packet: NymPacket::Sphinx(make_valid_sphinx_packet(Default::default())),
packet: Some(NymPacket::Sphinx(make_valid_sphinx_packet(
Default::default(),
))),
};
let mut bytes = BytesMut::new();
@@ -351,7 +387,7 @@ mod packet_encoding {
bytes.put_u8(PacketMode::default() as u8);
assert!(SphinxCodec.decode(&mut bytes).unwrap().is_some());
assert!(bytes.capacity() >= Header::LEGACY_SIZE + packet_size.size())
// assert!(bytes.capacity() >= Header::LEGACY_SIZE + packet_size.size())
}
}
@@ -369,7 +405,9 @@ mod packet_encoding {
for packet_size in packet_sizes {
let first_packet = FramedNymPacket {
header: Header::default(),
packet: NymPacket::Sphinx(make_valid_sphinx_packet(Default::default())),
packet: Some(NymPacket::Sphinx(make_valid_sphinx_packet(
Default::default(),
))),
};
let mut bytes = BytesMut::new();
@@ -379,7 +417,7 @@ mod packet_encoding {
bytes.put_u8(PacketMode::default() as u8);
assert!(SphinxCodec.decode(&mut bytes).unwrap().is_some());
assert!(bytes.capacity() >= Header::VERSIONED_SIZE + packet_size.size())
// assert!(bytes.capacity() >= Header::VERSIONED_SIZE + packet_size.size())
}
}
}
@@ -388,12 +426,16 @@ mod packet_encoding {
fn can_decode_two_packets_immediately() {
let packet1 = FramedNymPacket {
header: Header::default(),
packet: NymPacket::Sphinx(make_valid_sphinx_packet(Default::default())),
packet: Some(NymPacket::Sphinx(make_valid_sphinx_packet(
Default::default(),
))),
};
let packet2 = FramedNymPacket {
header: Header::default(),
packet: NymPacket::Sphinx(make_valid_sphinx_packet(Default::default())),
packet: Some(NymPacket::Sphinx(make_valid_sphinx_packet(
Default::default(),
))),
};
let mut bytes = BytesMut::new();
@@ -410,28 +452,38 @@ mod packet_encoding {
fn can_decode_two_packets_in_separate_calls() {
let packet1 = FramedNymPacket {
header: Header::default(),
packet: NymPacket::Sphinx(make_valid_sphinx_packet(Default::default())),
packet: Some(NymPacket::Sphinx(make_valid_sphinx_packet(
Default::default(),
))),
};
let packet2 = FramedNymPacket {
header: Header::default(),
packet: NymPacket::Sphinx(make_valid_sphinx_packet(Default::default())),
packet: Some(NymPacket::Sphinx(make_valid_sphinx_packet(
Default::default(),
))),
};
let mut bytes = BytesMut::new();
let mut bytes_tmp = BytesMut::new();
println!("{}", bytes.len());
SphinxCodec.encode(packet1, &mut bytes).unwrap();
println!("{}", bytes.len());
SphinxCodec.encode(packet2, &mut bytes_tmp).unwrap();
println!("{}", bytes.len());
let tmp = bytes_tmp.split_off(100);
bytes.put(bytes_tmp);
println!("{}", bytes.len());
assert!(SphinxCodec.decode(&mut bytes).unwrap().is_some());
println!("{}", bytes.len());
assert!(SphinxCodec.decode(&mut bytes).unwrap().is_none());
println!("{}", bytes.len());
bytes.put(tmp);
println!("{}", bytes.len());
assert!(SphinxCodec.decode(&mut bytes).unwrap().is_some());
assert!(SphinxCodec.decode(&mut bytes).unwrap().is_none());
}
common/nymsphinx/framing/src/packet.rs
+5 -5
View File
@@ -7,16 +7,16 @@ use nym_sphinx_params::packet_sizes::PacketSize;
use nym_sphinx_params::packet_version::PacketVersion;
use nym_sphinx_params::PacketMode;
use nym_sphinx_types::NymPacket;
use serde::{Serialize, Deserialize};
use serde::{Deserialize, Serialize};
use std::convert::TryFrom;
#[derive(Serialize, Deserialize)]
#[derive(Serialize, Deserialize, Debug)]
pub struct FramedNymPacket {
/// Contains any metadata helping receiver to handle the underlying packet.
pub(crate) header: Header,
/// The actual SphinxPacket being sent.
pub(crate) packet: NymPacket,
pub(crate) packet: Option<NymPacket>,
}
impl FramedNymPacket {
@@ -31,7 +31,7 @@ impl FramedNymPacket {
packet_size,
packet_mode,
},
packet,
packet: Some(packet),
}
}
@@ -48,7 +48,7 @@ impl FramedNymPacket {
}
pub fn into_inner(self) -> NymPacket {
self.packet
self.packet.unwrap()
}
}
common/nymsphinx/types/src/lib.rs
+15
View File
@@ -23,6 +23,21 @@ pub enum NymPacket {
Outfox(OutfoxPacket),
}
impl fmt::Debug for NymPacket {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match &self {
NymPacket::Sphinx(packet) => f
.debug_struct("NymPacket::Sphinx")
.field("len", &packet.len())
.finish(),
NymPacket::Outfox(packet) => f
.debug_struct("NymPacket::Outfox")
.field("len", &packet.len())
.finish(),
}
}
}
impl NymPacket {
pub fn len(&self) -> usize {
match self {