mab122
/
radicle-heartwood-lfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

node: Move over to binary protocol 

Signed-off-by: Alexis Sellier <self@cloudhead.io>
This commit is contained in:
Alexis Sellier 2022-09-08 14:36:09 +02:00
parent b3d3077ccc
commit 9d3030a093
No known key found for this signature in database
8 changed files with 386 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
node/src/decoder.rs
View File

@ -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<D> From<Vec<u8>> for Decoder<D> {
}
}
impl<'de, D: Deserialize<'de>> Decoder<D> {
impl<D: wire::Decode> Decoder<D> {
/// Create a new stream decoder.
pub fn new(capacity: usize) -> Self {
Self {
@ -36,23 +37,36 @@ impl<'de, D: Deserialize<'de>> Decoder<D> {
}
/// Decode and return the next message. Returns [`None`] if nothing was decoded.
pub fn decode_next(&mut self) -> Result<Option<D>, serde_json::Error> {
let mut de = serde_json::Deserializer::from_reader(self.unparsed.as_slice()).into_iter();
pub fn decode_next(&mut self) -> Result<Option<D>, 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<D> {
type Item = Result<D, serde_json::Error>;
impl<D: wire::Decode> io::Write for Decoder<D> {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
self.input(buf);
Ok(buf.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
impl<D: wire::Decode> Iterator for Decoder<D> {
type Item = Result<D, wire::Error>;
fn next(&mut self) -> Option<Self::Item> {
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::<serde_json::Value>::new(64);
let mut decoder = Decoder::<String>::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"));
}
}

26
node/src/protocol.rs
View File

@ -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::<Vec<_>>();
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<S, T, G> Context<S, T, G> {
}
fn write_all(&mut self, remote: net::SocketAddr, msgs: impl IntoIterator<Item = Message>) {
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));
}

7
node/src/protocol/config.rs
View File

@ -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()
},
}
}
}

87
node/src/protocol/message.rs
View File

@ -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<W: std::io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, std::io::Error> {
let mut n = 0;
n += self.magic.encode(writer)?;
n += self.msg.encode(writer)?;
Ok(n)
}
}
impl wire::Decode for Envelope {
fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
let magic = u32::decode(reader)?;
let msg = Message::decode(reader)?;
Ok(Self { magic, msg })
}
}
impl wire::Encode for Address {
fn encode<W: std::io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, std::io::Error> {
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::<Message>(&wire::serialize(&message)).unwrap(),
message
);
}
#[quickcheck]
fn prop_envelope_encode_decode(envelope: Envelope) {
assert_eq!(
wire::deserialize::<Envelope>(&wire::serialize(&envelope)).unwrap(),
envelope
);
}
#[test]
fn prop_envelope_decoder() {
fn property(items: Vec<Envelope>) {
let mut decoder = Decoder::<Envelope>::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<Envelope>));
}
#[quickcheck]
fn prop_addr(addr: Address) {
assert_eq!(
wire::deserialize::<Address>(&wire::serialize(&addr)).unwrap(),
addr
);
}
}

173
node/src/protocol/wire.rs
View File

@ -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<T: Encode + ?Sized>(data: &T) -> Vec<u8> {
buffer
}
/// Decode an object from a vector.
pub fn deserialize<T: Decode>(data: &[u8]) -> Result<T, Error> {
let mut cursor = io::Cursor::new(data);
T::decode(&mut cursor)
}
impl Encode for u8 {
fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
writer.write_u8(*self)?;
@ -79,13 +97,14 @@ impl Encode for u64 {
impl Encode for usize {
fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
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::<NetworkEndian>(*self as u32)?;
writer.write_u32::<NetworkEndian>(u)?;
Ok(mem::size_of_val(&u))
Ok(mem::size_of::<u32>())
}
}
@ -134,14 +153,18 @@ impl Encode for net::IpAddr {
}
}
impl Encode for str {
impl Encode for &str {
fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
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<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
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<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
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<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
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::<u8>(&serialize(&input)).unwrap(), input);
}
#[quickcheck]
fn prop_u16(input: u16) {
assert_eq!(deserialize::<u16>(&serialize(&input)).unwrap(), input);
}
#[quickcheck]
fn prop_u32(input: u32) {
assert_eq!(deserialize::<u32>(&serialize(&input)).unwrap(), input);
}
#[quickcheck]
fn prop_u64(input: u64) {
assert_eq!(deserialize::<u64>(&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::<usize>(&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::<String>(&serialize(&input)).unwrap(), input);
quickcheck::TestResult::passed()
}
#[quickcheck]
fn prop_pubkey(input: PublicKey) {
assert_eq!(deserialize::<PublicKey>(&serialize(&input)).unwrap(), input);
}
#[quickcheck]
fn prop_id(input: Id) {
assert_eq!(deserialize::<Id>(&serialize(&input)).unwrap(), input);
}
#[quickcheck]
fn prop_digest(input: Digest) {
assert_eq!(deserialize::<Digest>(&serialize(&input)).unwrap(), input);
}
#[quickcheck]
fn prop_refs(input: Refs) {
assert_eq!(deserialize::<Refs>(&serialize(&input)).unwrap(), input);
}
#[quickcheck]
fn prop_signature(input: arbitrary::ByteArray<64>) {
let signature = Signature::from(input.into_inner());
assert_eq!(
deserialize::<Signature>(&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::<git::Oid>(&serialize(&oid)).unwrap(), oid);
}
#[quickcheck]
fn prop_signed_refs(input: SignedRefs<Unverified>) {
assert_eq!(
deserialize::<SignedRefs<Unverified>>(&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::<git::Url>(&serialize(&url)).unwrap(), url);
}
}

3
node/src/storage/git.rs
View File

@ -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()
}
}

102
node/src/test/arbitrary.rs
View File

@ -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<T: Arbitrary>(size: usize) -> T {
T::arbitrary(&mut gen)
}
#[derive(Clone, Debug)]
pub struct ByteArray<const N: usize>([u8; N]);
impl<const N: usize> ByteArray<N> {
pub fn into_inner(self) -> [u8; N] {
self.0
}
}
impl<const N: usize> Arbitrary for ByteArray<N> {
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::<Id>::arbitrary(g),
},
6 => Self::Inventory {
node: NodeId::arbitrary(g),
inv: Vec::<Id>::arbitrary(g),
timestamp: Timestamp::arbitrary(g),
},
8 => Self::RefsUpdate {
id: Id::arbitrary(g),
signer: PublicKey::arbitrary(g),
refs: SignedRefs::<Unverified>::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<crypto::Verified> {
fn arbitrary(g: &mut quickcheck::Gen) -> Self {
let remotes: HashMap<storage::RemoteId, storage::Remote<crypto::Verified>> =
@ -104,6 +176,16 @@ impl Arbitrary for Doc {
}
}
impl Arbitrary for SignedRefs<Unverified> {
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<git::RefString, storage::Oid> = BTreeMap::new();
@ -146,12 +228,8 @@ impl Arbitrary for storage::Remote<crypto::Verified> {
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)

4
node/src/test/peer.rs
View File

@ -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::<Protocol<S>>::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();