From 9d3030a0933aaf96125e1e5ea1683915ff867e2a Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Thu, 8 Sep 2022 14:36:09 +0200 Subject: [PATCH] node: Move over to binary protocol Signed-off-by: Alexis Sellier --- node/src/decoder.rs | 58 ++++++------ node/src/protocol.rs | 26 ++++-- node/src/protocol/config.rs | 7 +- node/src/protocol/message.rs | 87 +++++++++++++++++- node/src/protocol/wire.rs | 173 +++++++++++++++++++++++++++++++---- node/src/storage/git.rs | 3 +- node/src/test/arbitrary.rs | 102 ++++++++++++++++++--- node/src/test/peer.rs | 4 +- 8 files changed, 386 insertions(+), 74 deletions(-) diff --git a/node/src/decoder.rs b/node/src/decoder.rs index 66d7b95c..3b786093 100644 --- a/node/src/decoder.rs +++ b/node/src/decoder.rs @@ -1,7 +1,8 @@ +use std::io; use std::marker::PhantomData; use crate::protocol::message::Envelope; -use serde::Deserialize; +use crate::protocol::wire; /// Message stream decoder. /// @@ -21,7 +22,7 @@ impl From> for Decoder { } } -impl<'de, D: Deserialize<'de>> Decoder { +impl Decoder { /// Create a new stream decoder. pub fn new(capacity: usize) -> Self { Self { @@ -36,23 +37,36 @@ impl<'de, D: Deserialize<'de>> Decoder { } /// Decode and return the next message. Returns [`None`] if nothing was decoded. - pub fn decode_next(&mut self) -> Result, serde_json::Error> { - let mut de = serde_json::Deserializer::from_reader(self.unparsed.as_slice()).into_iter(); + pub fn decode_next(&mut self) -> Result, wire::Error> { + let mut reader = io::Cursor::new(self.unparsed.as_mut_slice()); + + match D::decode(&mut reader) { + Ok(msg) => { + let pos = reader.position() as usize; + self.unparsed.drain(..pos); - match de.next() { - Some(Ok(msg)) => { - self.unparsed.drain(..de.byte_offset()); Ok(Some(msg)) } - Some(Err(err)) if err.is_eof() => Ok(None), - - result => result.transpose(), + Err(err) if err.is_eof() => Ok(None), + Err(err) => Err(err), } } } -impl<'de, D: Deserialize<'de>> Iterator for Decoder { - type Item = Result; +impl io::Write for Decoder { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.input(buf); + + Ok(buf.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +impl Iterator for Decoder { + type Item = Result; fn next(&mut self) -> Option { self.decode_next().transpose() @@ -64,14 +78,14 @@ mod test { use super::*; use quickcheck_macros::quickcheck; - const MSG_HELLO: &[u8] = b"{\"cmd\":\"hello\"}"; - const MSG_BYE: &[u8] = b"{\"cmd\":\"goodbye\"}"; + const MSG_HELLO: &[u8] = &[5, b'h', b'e', b'l', b'l', b'o']; + const MSG_BYE: &[u8] = &[3, b'b', b'y', b'e']; #[quickcheck] fn prop_decode_next(chunk_size: usize) { let mut bytes = vec![]; let mut msgs = vec![]; - let mut decoder = Decoder::::new(64); + let mut decoder = Decoder::::new(8); let chunk_size = 1 + chunk_size % MSG_HELLO.len() + MSG_BYE.len(); @@ -88,17 +102,7 @@ mod test { assert_eq!(decoder.unparsed.len(), 0); assert_eq!(msgs.len(), 2); - assert_eq!( - msgs[0], - serde_json::json!({ - "cmd": "hello", - }) - ); - assert_eq!( - msgs[1], - serde_json::json!({ - "cmd": "goodbye", - }) - ); + assert_eq!(msgs[0], String::from("hello")); + assert_eq!(msgs[1], String::from("bye")); } } diff --git a/node/src/protocol.rs b/node/src/protocol.rs index 95f58b91..6712e55d 100644 --- a/node/src/protocol.rs +++ b/node/src/protocol.rs @@ -5,7 +5,7 @@ pub mod peer; pub mod wire; use std::ops::{Deref, DerefMut}; -use std::{collections::VecDeque, fmt, io, net, net::IpAddr}; +use std::{collections::VecDeque, fmt, net, net::IpAddr}; use crossbeam_channel as chan; use fastrand::Rng; @@ -26,6 +26,7 @@ use crate::identity::{Id, Project, PublicKey}; use crate::protocol::config::ProjectTracking; use crate::protocol::message::Message; use crate::protocol::peer::{Peer, PeerError, PeerState}; +use crate::protocol::wire::Encode; use crate::storage::{self, ReadRepository, WriteRepository}; use crate::storage::{Inventory, WriteStorage}; @@ -546,7 +547,7 @@ where .collect::>(); let (peer, msgs) = if let Some(peer) = self.peers.get_mut(&peer) { - let decoder = &mut peer.inbox(); + let decoder = peer.inbox(); decoder.input(bytes); let mut msgs = Vec::with_capacity(1); @@ -555,8 +556,10 @@ where Ok(Some(msg)) => msgs.push(msg), Ok(None) => break, - Err(_err) => { + Err(err) => { // TODO: Disconnect peer. + error!("Invalid message received from {}: {}", peer.addr, err); + return; } } @@ -731,8 +734,13 @@ where fn fetch(&mut self, proj_id: &Id, remote: &Url) { // TODO: Verify refs before adding them to storage. let mut repo = self.storage.repository(proj_id).unwrap(); + let mut path = remote.path.clone(); + + path.push(b'/'); + path.extend(proj_id.to_string().into_bytes()); + repo.fetch(&Url { - path: format!("/{}", proj_id).into(), + path, ..remote.clone() }) .unwrap(); @@ -757,22 +765,24 @@ impl Context { } fn write_all(&mut self, remote: net::SocketAddr, msgs: impl IntoIterator) { - let mut buf = io::Cursor::new(Vec::new()); + let mut buf = Vec::new(); for msg in msgs { debug!("Write {:?} to {}", &msg, remote.ip()); let envelope = self.config.network.envelope(msg); - serde_json::to_writer(&mut buf, &envelope).unwrap(); + envelope + .encode(&mut buf) + .expect("writing to an in-memory buffer doesn't fail"); } - self.io.push_back(Io::Write(remote, buf.into_inner())); + self.io.push_back(Io::Write(remote, buf)); } fn write(&mut self, remote: net::SocketAddr, msg: Message) { debug!("Write {:?} to {}", &msg, remote.ip()); let envelope = self.config.network.envelope(msg); - let bytes = serde_json::to_vec(&envelope).unwrap(); + let bytes = wire::serialize(&envelope); self.io.push_back(Io::Write(remote, bytes)); } diff --git a/node/src/protocol/config.rs b/node/src/protocol/config.rs index f6a6177d..d2bfc62c 100644 --- a/node/src/protocol/config.rs +++ b/node/src/protocol/config.rs @@ -3,6 +3,7 @@ use std::net; use git_url::Url; use crate::collections::HashSet; +use crate::git; use crate::identity::{Id, PublicKey}; use crate::protocol::message::{Address, Envelope, Message}; @@ -88,7 +89,11 @@ impl Default for Config { remote_tracking: RemoteTracking::default(), relay: true, listen: vec![], - git_url: Url::default(), + git_url: Url { + scheme: git::url::Scheme::File, + path: "/dev/null".to_owned().into(), + ..Url::default() + }, } } } diff --git a/node/src/protocol/message.rs b/node/src/protocol/message.rs index 5f9e00ae..50d57a69 100644 --- a/node/src/protocol/message.rs +++ b/node/src/protocol/message.rs @@ -12,7 +12,7 @@ use crate::storage; use crate::storage::refs::SignedRefs; /// Message envelope. All messages sent over the network are wrapped in this type. -#[derive(Debug, Serialize, Deserialize)] +#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)] pub struct Envelope { /// Network magic constant. Used to differentiate networks. pub magic: u32, @@ -47,6 +47,26 @@ pub enum Address { }, } +impl wire::Encode for Envelope { + fn encode(&self, writer: &mut W) -> Result { + let mut n = 0; + + n += self.magic.encode(writer)?; + n += self.msg.encode(writer)?; + + Ok(n) + } +} + +impl wire::Decode for Envelope { + fn decode(reader: &mut R) -> Result { + let magic = u32::decode(reader)?; + let msg = Message::decode(reader)?; + + Ok(Self { magic, msg }) + } +} + impl wire::Encode for Address { fn encode(&self, writer: &mut W) -> Result { let mut n = 0; @@ -84,6 +104,13 @@ impl wire::Decode for Address { Ok(Self::Ip { ip, port }) } + 2 => { + let octets: [u8; 16] = wire::Decode::decode(reader)?; + let ip = net::IpAddr::from(net::Ipv6Addr::from(octets)); + let port = u16::decode(reader)?; + + Ok(Self::Ip { ip, port }) + } _ => { todo!(); } @@ -108,6 +135,7 @@ pub struct NodeAnnouncement { impl NodeAnnouncement { /// Verify a signature on this message. pub fn verify(&self, signature: &crypto::Signature) -> bool { + // TODO: Use binary serialization. let msg = serde_json::to_vec(self).unwrap(); self.id.verify(signature, &msg).is_ok() } @@ -115,7 +143,7 @@ impl NodeAnnouncement { /// Message payload. /// These are the messages peers send to each other. -#[derive(Debug, Serialize, Deserialize, Clone)] +#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)] pub enum Message { /// Say hello to a peer. This is the first message sent to a peer after connection. Hello { @@ -295,9 +323,60 @@ impl wire::Decode for Message { Ok(Self::RefsUpdate { id, signer, refs }) } - _ => { - todo!(); + n => { + todo!("Mesage type {} is not yet implemented", n); } } } } + +#[cfg(test)] +mod tests { + use super::*; + use quickcheck_macros::quickcheck; + + use crate::decoder::Decoder; + use crate::protocol::wire::{self, Encode}; + + #[quickcheck] + fn prop_message_encode_decode(message: Message) { + assert_eq!( + wire::deserialize::(&wire::serialize(&message)).unwrap(), + message + ); + } + + #[quickcheck] + fn prop_envelope_encode_decode(envelope: Envelope) { + assert_eq!( + wire::deserialize::(&wire::serialize(&envelope)).unwrap(), + envelope + ); + } + + #[test] + fn prop_envelope_decoder() { + fn property(items: Vec) { + let mut decoder = Decoder::::new(8); + + for item in &items { + item.encode(&mut decoder).unwrap(); + } + for item in items { + assert_eq!(decoder.next().unwrap().unwrap(), item); + } + } + + quickcheck::QuickCheck::new() + .gen(quickcheck::Gen::new(16)) + .quickcheck(property as fn(items: Vec)); + } + + #[quickcheck] + fn prop_addr(addr: Address) { + assert_eq!( + wire::deserialize::
(&wire::serialize(&addr)).unwrap(), + addr + ); + } +} diff --git a/node/src/protocol/wire.rs b/node/src/protocol/wire.rs index 07bfb08b..593741cd 100644 --- a/node/src/protocol/wire.rs +++ b/node/src/protocol/wire.rs @@ -23,6 +23,17 @@ pub enum Error { InvalidSize { expected: usize, actual: usize }, #[error(transparent)] InvalidRefName(#[from] fmt::Error), + #[error("invalid git url `{url}`: {error}")] + InvalidGitUrl { + url: String, + error: git::url::parse::Error, + }, +} + +impl Error { + pub fn is_eof(&self) -> bool { + matches!(self, Self::Io(err) if err.kind() == io::ErrorKind::UnexpectedEof) + } } pub trait Encode { @@ -45,6 +56,13 @@ pub fn serialize(data: &T) -> Vec { buffer } +/// Decode an object from a vector. +pub fn deserialize(data: &[u8]) -> Result { + let mut cursor = io::Cursor::new(data); + + T::decode(&mut cursor) +} + impl Encode for u8 { fn encode(&self, writer: &mut W) -> Result { writer.write_u8(*self)?; @@ -79,13 +97,14 @@ impl Encode for u64 { impl Encode for usize { fn encode(&self, writer: &mut W) -> Result { - let u = (*self) - .try_into() - .map_err(|_| io::Error::from(io::ErrorKind::InvalidInput))?; + assert!( + *self <= u32::MAX as usize, + "Cannot encode sizes larger than {}", + u32::MAX + ); + writer.write_u32::(*self as u32)?; - writer.write_u32::(u)?; - - Ok(mem::size_of_val(&u)) + Ok(mem::size_of::()) } } @@ -134,14 +153,18 @@ impl Encode for net::IpAddr { } } -impl Encode for str { +impl Encode for &str { fn encode(&self, writer: &mut W) -> Result { - let mut n = 0; + assert!(self.len() <= u8::MAX as usize); - n += self.len().encode(writer)?; - n += self.as_bytes().encode(writer)?; + let n = (self.len() as u8).encode(writer)?; + let bytes = self.as_bytes(); - Ok(n) + // Nb. Don't use the [`Encode`] instance here for &[u8], because we are prefixing the + // length ourselves. + writer.write_all(bytes)?; + + Ok(n + bytes.len()) } } @@ -276,7 +299,7 @@ impl Decode for u64 { impl Decode for usize { fn decode(reader: &mut R) -> Result { - let size: usize = u64::decode(reader)? + let size: usize = u32::decode(reader)? .try_into() .map_err(|_| io::Error::from(io::ErrorKind::InvalidInput))?; @@ -311,8 +334,8 @@ where impl Decode for String { fn decode(reader: &mut R) -> Result { - let len = usize::decode(reader)?; - let mut bytes = vec![0; len]; + let len = u8::decode(reader)?; + let mut bytes = vec![0; len as usize]; reader.read_exact(&mut bytes)?; @@ -324,10 +347,11 @@ impl Decode for String { impl Decode for git::Url { fn decode(reader: &mut R) -> Result { - let string = String::decode(reader)?; + let url = String::decode(reader)?; + let url = Self::from_bytes(url.as_bytes()) + .map_err(|error| Error::InvalidGitUrl { url, error })?; - Self::from_bytes(string.as_bytes()) - .map_err(|e| Error::Io(io::Error::new(io::ErrorKind::InvalidInput, e.to_string()))) + Ok(url) } } @@ -346,3 +370,118 @@ impl Decode for Digest { Ok(Self::from(bytes)) } } + +#[cfg(test)] +mod tests { + use super::*; + use quickcheck_macros::quickcheck; + + use crate::crypto::Unverified; + use crate::storage::refs::SignedRefs; + use crate::test::arbitrary; + + #[quickcheck] + fn prop_u8(input: u8) { + assert_eq!(deserialize::(&serialize(&input)).unwrap(), input); + } + + #[quickcheck] + fn prop_u16(input: u16) { + assert_eq!(deserialize::(&serialize(&input)).unwrap(), input); + } + + #[quickcheck] + fn prop_u32(input: u32) { + assert_eq!(deserialize::(&serialize(&input)).unwrap(), input); + } + + #[quickcheck] + fn prop_u64(input: u64) { + assert_eq!(deserialize::(&serialize(&input)).unwrap(), input); + } + + #[quickcheck] + fn prop_usize(input: usize) -> quickcheck::TestResult { + if input > u32::MAX as usize { + return quickcheck::TestResult::discard(); + } + assert_eq!(deserialize::(&serialize(&input)).unwrap(), input); + + quickcheck::TestResult::passed() + } + + #[quickcheck] + fn prop_string(input: String) -> quickcheck::TestResult { + if input.len() > u8::MAX as usize { + return quickcheck::TestResult::discard(); + } + assert_eq!(deserialize::(&serialize(&input)).unwrap(), input); + + quickcheck::TestResult::passed() + } + + #[quickcheck] + fn prop_pubkey(input: PublicKey) { + assert_eq!(deserialize::(&serialize(&input)).unwrap(), input); + } + + #[quickcheck] + fn prop_id(input: Id) { + assert_eq!(deserialize::(&serialize(&input)).unwrap(), input); + } + + #[quickcheck] + fn prop_digest(input: Digest) { + assert_eq!(deserialize::(&serialize(&input)).unwrap(), input); + } + + #[quickcheck] + fn prop_refs(input: Refs) { + assert_eq!(deserialize::(&serialize(&input)).unwrap(), input); + } + + #[quickcheck] + fn prop_signature(input: arbitrary::ByteArray<64>) { + let signature = Signature::from(input.into_inner()); + + assert_eq!( + deserialize::(&serialize(&signature)).unwrap(), + signature + ); + } + + #[quickcheck] + fn prop_oid(input: arbitrary::ByteArray<20>) { + let oid = git::Oid::try_from(input.into_inner().as_slice()).unwrap(); + + assert_eq!(deserialize::(&serialize(&oid)).unwrap(), oid); + } + + #[quickcheck] + fn prop_signed_refs(input: SignedRefs) { + assert_eq!( + deserialize::>(&serialize(&input)).unwrap(), + input + ); + } + + #[test] + fn test_string() { + assert_eq!( + serialize(&String::from("hello")), + vec![5, b'h', b'e', b'l', b'l', b'o'] + ); + } + + #[test] + fn test_git_url() { + let url = git::Url { + scheme: git::url::Scheme::Https, + path: "/git".to_owned().into(), + host: Some("seed.radicle.xyz".to_owned()), + port: Some(8888), + ..git::Url::default() + }; + assert_eq!(deserialize::(&serialize(&url)).unwrap(), url); + } +} diff --git a/node/src/storage/git.rs b/node/src/storage/git.rs index e2e67bda..a86ea3dc 100644 --- a/node/src/storage/git.rs +++ b/node/src/storage/git.rs @@ -45,7 +45,8 @@ impl ReadStorage for Storage { fn url(&self) -> git::Url { git::Url { scheme: git_url::Scheme::File, - host: Some(self.path.to_string_lossy().to_string()), + host: None, + path: self.path.to_string_lossy().to_string().into(), ..git::Url::default() } } diff --git a/node/src/test/arbitrary.rs b/node/src/test/arbitrary.rs index 5f141dc0..05f554f8 100644 --- a/node/src/test/arbitrary.rs +++ b/node/src/test/arbitrary.rs @@ -1,5 +1,6 @@ use std::collections::{BTreeMap, HashSet}; use std::hash::Hash; +use std::net; use std::ops::RangeBounds; use std::path::PathBuf; @@ -7,13 +8,15 @@ use nonempty::NonEmpty; use quickcheck::Arbitrary; use crate::collections::HashMap; -use crate::crypto::{self, Signer}; +use crate::crypto::{self, Signer, Unverified}; use crate::crypto::{PublicKey, SecretKey}; use crate::git; use crate::hash; use crate::identity::{Delegate, Did, Doc, Id, Project}; +use crate::protocol::message::{Address, Envelope, Message}; +use crate::protocol::{NodeId, Timestamp}; use crate::storage; -use crate::storage::refs::Refs; +use crate::storage::refs::{Refs, SignedRefs}; use crate::test::storage::MockStorage; use super::crypto::MockSigner; @@ -35,6 +38,75 @@ pub fn gen(size: usize) -> T { T::arbitrary(&mut gen) } +#[derive(Clone, Debug)] +pub struct ByteArray([u8; N]); + +impl ByteArray { + pub fn into_inner(self) -> [u8; N] { + self.0 + } +} + +impl Arbitrary for ByteArray { + fn arbitrary(g: &mut quickcheck::Gen) -> Self { + let mut bytes: [u8; N] = [0; N]; + for byte in &mut bytes { + *byte = u8::arbitrary(g); + } + Self(bytes) + } +} + +impl Arbitrary for Envelope { + fn arbitrary(g: &mut quickcheck::Gen) -> Self { + Self { + magic: u32::arbitrary(g), + msg: Message::arbitrary(g), + } + } +} + +impl Arbitrary for Message { + fn arbitrary(g: &mut quickcheck::Gen) -> Self { + let type_id = g.choose(&[4, 6, 8]).unwrap(); + + match type_id { + 4 => Self::GetInventory { + ids: Vec::::arbitrary(g), + }, + 6 => Self::Inventory { + node: NodeId::arbitrary(g), + inv: Vec::::arbitrary(g), + timestamp: Timestamp::arbitrary(g), + }, + 8 => Self::RefsUpdate { + id: Id::arbitrary(g), + signer: PublicKey::arbitrary(g), + refs: SignedRefs::::arbitrary(g), + }, + _ => unreachable!(), + } + } +} + +impl Arbitrary for Address { + fn arbitrary(g: &mut quickcheck::Gen) -> Self { + if bool::arbitrary(g) { + Address::Ip { + ip: net::IpAddr::V4(net::Ipv4Addr::from(u32::arbitrary(g))), + port: u16::arbitrary(g), + } + } else { + let octets: [u8; 16] = ByteArray::<16>::arbitrary(g).into_inner(); + + Address::Ip { + ip: net::IpAddr::V6(net::Ipv6Addr::from(octets)), + port: u16::arbitrary(g), + } + } + } +} + impl Arbitrary for storage::Remotes { fn arbitrary(g: &mut quickcheck::Gen) -> Self { let remotes: HashMap> = @@ -104,6 +176,16 @@ impl Arbitrary for Doc { } } +impl Arbitrary for SignedRefs { + fn arbitrary(g: &mut quickcheck::Gen) -> Self { + let bytes: ByteArray<64> = Arbitrary::arbitrary(g); + let signature = crypto::Signature::from(bytes.into_inner()); + let refs = Refs::arbitrary(g); + + Self::new(refs, signature) + } +} + impl Arbitrary for Refs { fn arbitrary(g: &mut quickcheck::Gen) -> Self { let mut refs: BTreeMap = BTreeMap::new(); @@ -146,12 +228,8 @@ impl Arbitrary for storage::Remote { impl Arbitrary for MockSigner { fn arbitrary(g: &mut quickcheck::Gen) -> Self { - let mut bytes: [u8; 32] = [0; 32]; - - for byte in &mut bytes { - *byte = u8::arbitrary(g); - } - MockSigner::from(SecretKey::from(bytes)) + let bytes: ByteArray<32> = Arbitrary::arbitrary(g); + MockSigner::from(SecretKey::from(bytes.into_inner())) } } @@ -173,12 +251,8 @@ impl Arbitrary for PublicKey { fn arbitrary(g: &mut quickcheck::Gen) -> Self { use ed25519_consensus::SigningKey; - let mut bytes: [u8; 32] = [0; 32]; - - for byte in &mut bytes { - *byte = u8::arbitrary(g); - } - let sk = SigningKey::from(bytes); + let bytes: ByteArray<32> = Arbitrary::arbitrary(g); + let sk = SigningKey::from(bytes.into_inner()); let vk = sk.verification_key(); PublicKey(vk) diff --git a/node/src/test/peer.rs b/node/src/test/peer.rs index 07285d2f..28c391d9 100644 --- a/node/src/test/peer.rs +++ b/node/src/test/peer.rs @@ -129,7 +129,7 @@ where } pub fn receive(&mut self, peer: &net::SocketAddr, msg: Message) { - let bytes = serde_json::to_vec(&self.config().network.envelope(msg)).unwrap(); + let bytes = wire::serialize(&self.config().network.envelope(msg)); self.protocol.received_bytes(peer, &bytes); } @@ -137,7 +137,7 @@ where pub fn connect_from(&mut self, peer: &Self) { let remote = simulator::Peer::>::addr(peer); let local = net::SocketAddr::new(self.ip, self.rng.u16(..)); - let git = format!("file://{}.git", remote.ip()); + let git = format!("file:///{}.git", remote.ip()); let git = Url::from_bytes(git.as_bytes()).unwrap(); self.initialize();