From 3e4854874574f733e181a2b4eada43b0adfeee01 Mon Sep 17 00:00:00 2001 From: "Dr. Maxim Orlovsky" Date: Mon, 14 Nov 2022 22:55:50 +0100 Subject: [PATCH] Integrate framing and muxer into wire protocol --- radicle-node/src/client.rs | 2 +- radicle-node/src/wire.rs | 45 ++++++++++++------- .../src/wire/{transcoder.rs => transcode.rs} | 31 ++++++++++++- 3 files changed, 61 insertions(+), 17 deletions(-) rename radicle-node/src/wire/{transcoder.rs => transcode.rs} (89%) diff --git a/radicle-node/src/client.rs b/radicle-node/src/client.rs index 938a5ca6..49abd3f0 100644 --- a/radicle-node/src/client.rs +++ b/radicle-node/src/client.rs @@ -9,7 +9,7 @@ use radicle::crypto::Signer; use crate::clock::RefClock; use crate::profile::Profile; use crate::service::routing; -use crate::wire::transcoder::NoHandshake; +use crate::wire::transcode::NoHandshake; use crate::wire::Wire; use crate::{address, service}; diff --git a/radicle-node/src/wire.rs b/radicle-node/src/wire.rs index 1204caa1..e5ad954b 100644 --- a/radicle-node/src/wire.rs +++ b/radicle-node/src/wire.rs @@ -1,5 +1,5 @@ pub mod message; -pub mod transcoder; +pub mod transcode; use std::collections::{BTreeMap, HashMap, VecDeque}; use std::convert::TryFrom; @@ -26,7 +26,7 @@ use crate::service::{filter, routing, session}; use crate::storage::refs::Refs; use crate::storage::refs::SignedRefs; use crate::storage::WriteStorage; -use crate::wire::transcoder::{Handshake, HandshakeResult, Transcode}; +use crate::wire::transcode::{Framer, Handshake, HandshakeResult, MuxMsg, Transcode}; /// The default type we use to represent sizes on the wire. /// @@ -426,7 +426,7 @@ impl Decode for node::Features { #[derive(Debug)] pub struct Inbox { - pub transcoder: T, + pub pipeline: Framer, pub deserializer: Deserializer, } @@ -515,10 +515,11 @@ where self.inner_queue .push_back(nakamoto::Io::Write(*addr, reply)); } + let pipeline = Framer::new(transcoder); self.inboxes.insert( *addr, Inbox { - transcoder, + pipeline, deserializer: Deserializer::new(256), }, ); @@ -538,18 +539,31 @@ where } if let Some(Inbox { - transcoder, + pipeline, deserializer, }) = self.inboxes.get_mut(addr) { - let bytes = transcoder.decode(raw_bytes); - deserializer.input(&bytes); - - loop { - match deserializer.deserialize_next() { - Ok(Some(msg)) => self.inner.received_message(addr, msg), - Ok(None) => break, + pipeline.input(raw_bytes); + for frame in pipeline { + let Ok(msg) = MuxMsg::try_from(frame) else { + // TODO: Disconnect peer. + log::error!("Message frame with invalid channel structure from {}", addr); + return; + }; + match msg.channel { + 0 => deserializer.input(&msg.data), + 1 => { /* TODO: Send to git worker */ } + wrong_channel => { + // TODO: Disconnect peer. + log::error!("Wrong message channel {} from peer {}", wrong_channel, addr); + return; + } + }; + } + for message in deserializer { + match message { + Ok(msg) => self.inner.received_message(addr, msg), Err(err) => { // TODO: Disconnect peer. log::error!("Invalid message received from {}: {}", addr, err); @@ -581,11 +595,12 @@ impl Iterator for Wire { msg.encode(&mut buf) .expect("writing to an in-memory buffer doesn't fail"); } - let Inbox { transcoder, .. } = self.inboxes.get_mut(&addr).expect( + let Inbox { pipeline, .. } = self.inboxes.get_mut(&addr).expect( "broken handshake implementation: data sent before handshake was complete", ); - let data = transcoder.encode(buf); - Some(nakamoto::Io::Write(addr, data)) + let data = pipeline.frame(buf).expect("oversized data for a frame"); + let msg = MuxMsg { channel: 0, data }; + Some(nakamoto::Io::Write(addr, msg.into())) } Some(Io::Event(e)) => Some(nakamoto::Io::Event(e)), Some(Io::Connect(a)) => Some(nakamoto::Io::Connect(a)), diff --git a/radicle-node/src/wire/transcoder.rs b/radicle-node/src/wire/transcode.rs similarity index 89% rename from radicle-node/src/wire/transcoder.rs rename to radicle-node/src/wire/transcode.rs index 2f25920a..da3e9cc5 100644 --- a/radicle-node/src/wire/transcoder.rs +++ b/radicle-node/src/wire/transcode.rs @@ -107,8 +107,10 @@ impl Framer { let len = u8::try_from(len).map_err(|_| OversizedData(len))?; let len = len.to_be_bytes(); let mut buf = Vec::with_capacity(data.len() + 2); + buf.extend(len); buf.append(&mut data); + Ok(buf) } } @@ -124,16 +126,20 @@ impl Iterator for Framer { self.input .read_exact(&mut len) .expect("the length is checked"); + let len = u16::from_be_bytes(len) as usize; if self.input.len() < 2 + len { return None; } self.input.pop_front(); self.input.pop_front(); + let reminder = self.input.split_off(len); let mut data = vec![0u8; len]; + self.input.read_exact(&mut data).expect("checked length"); self.input = reminder; + Some(data) } } @@ -147,6 +153,17 @@ pub struct MuxMsg { pub data: Vec, } +impl From for Frame { + fn from(mut msg: MuxMsg) -> Self { + let channel = msg.channel.to_be_bytes(); + let mut data = Vec::with_capacity(msg.data.len() + 2); + + data.extend(channel); + data.append(&mut msg.data); + data + } +} + impl TryFrom for MuxMsg { type Error = ChannelError; @@ -156,10 +173,13 @@ impl TryFrom for MuxMsg { } let mut channel = [0u8; 2]; let mut cursor = io::Cursor::new(frame); + cursor .read_exact(&mut channel) .expect("the length is checked"); + let channel = u16::from_be_bytes(channel); + Ok(MuxMsg { channel, data: cursor.into_inner(), @@ -170,16 +190,19 @@ impl TryFrom for MuxMsg { #[cfg(test)] mod test { use super::*; + use crate::deserializer::Deserializer; #[test] - fn transcode() { + fn decode() { let mut pipeline = Framer::new(PlainTranscoder); + let mut deser = Deserializer::::new(512); let data = [ 0x00, 0x04, 0x00, 0x00, b'a', b'b', 0x00, 0x07, 0x00, 0x01, b'M', b'a', b'x', b'i', b'm', ]; let mut expected_payloads = [(0u16, b"ab".to_vec()), (1, b"Maxim".to_vec())].into_iter(); + let mut expected_msgs = ["ab", "Maxim"].into_iter(); for byte in data { // Writing data byte by byte, ensuring that the reading is not broken @@ -187,8 +210,14 @@ mod test { for frame in &mut pipeline { let msg = MuxMsg::try_from(frame).unwrap(); let (channel, data) = expected_payloads.next().unwrap(); + deser.input(&data); assert_eq!(msg, MuxMsg { channel, data }); } } + + for msg in deser { + let msg = msg.unwrap(); + assert_eq!(msg, expected_msgs.next().unwrap()); + } } }