From c4b3f4c3ea2005187b085093e2cffcb9c1c3d2fa Mon Sep 17 00:00:00 2001 From: Lee Smet Date: Tue, 3 Oct 2023 10:14:53 +0200 Subject: [PATCH] Initial message structure Signed-off-by: Lee Smet --- src/data.rs | 147 ++++++++++++++++++++++------------- src/lib.rs | 1 + src/main.rs | 9 ++- src/message.rs | 206 +++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 308 insertions(+), 55 deletions(-) create mode 100644 src/message.rs diff --git a/src/data.rs b/src/data.rs index 52cdba7..7dd77c5 100644 --- a/src/data.rs +++ b/src/data.rs @@ -1,4 +1,4 @@ -use std::{marker::PhantomData, net::Ipv6Addr}; +use std::net::Ipv6Addr; use futures::{Sink, SinkExt, Stream, StreamExt}; use log::{debug, error, trace, warn}; @@ -12,6 +12,9 @@ const USER_DATA_VERSION: u8 = 1; /// Type value indicating L3 data in the user data header. const USER_DATA_L3_TYPE: u8 = 0; +/// Type value indicating a user message in the data header. +const USER_DATA_MESSAGE_TYPE: u8 = 1; + /// Minimum size in bytes of an IPv6 header. const IPV6_MIN_HEADER_SIZE: usize = 40; @@ -24,44 +27,42 @@ const IPV6_VERSION_BYTE: u8 = 0b0110_0000; /// The DataPlane manages forwarding/receiving of local data packets to the [`Router`], and the /// encryption/decryption of them. +/// +/// DataPlane itself can be cloned, but this is not cheap on the router and should be avoided. #[derive(Clone)] -pub struct DataPlane { +pub struct DataPlane { router: Router, - _stream_marker: PhantomData, - _sink_marker: PhantomData, } -impl DataPlane -where - S: Stream> + Send + Unpin + 'static, - T: Sink + Send + Unpin + 'static, - T::Error: std::fmt::Display, -{ +impl DataPlane { /// Create a new `DataPlane` using the given [`Router`] for packet handling. /// /// `l3_packet_stream` is a stream of l3 packets from the host, usually read from a TUN interface. /// `l3_packet_sink` is a sink for l3 packets received from a romte, usually send to a TUN interface, - pub fn new( + pub fn new( router: Router, l3_packet_stream: S, l3_packet_sink: T, + message_packet_sink: U, host_packet_source: UnboundedReceiver, - ) -> Self { - tokio::spawn(Self::inject_l3_packet_loop( - router.clone(), - l3_packet_stream, - )); - tokio::spawn(Self::extract_l3_packet_loop( - router.clone(), + ) -> Self + where + S: Stream> + Send + Unpin + 'static, + T: Sink + Send + Unpin + 'static, + T::Error: std::fmt::Display, + U: Sink + Send + Unpin + 'static, + U::Error: std::fmt::Display, + { + let dp = Self { router }; + + tokio::spawn(dp.clone().inject_l3_packet_loop(l3_packet_stream)); + tokio::spawn(dp.clone().extract_packet_loop( l3_packet_sink, + message_packet_sink, host_packet_source, )); - Self { - router, - _stream_marker: PhantomData, - _sink_marker: PhantomData, - } + dp } /// Get a reference to the [`Router`] used. @@ -69,7 +70,12 @@ where &self.router } - async fn inject_l3_packet_loop(router: Router, mut l3_packet_stream: S) { + async fn inject_l3_packet_loop(self, mut l3_packet_stream: S) + where + // TODO: no result + // TODO: should IP extraction be handled higher up? + S: Stream> + Send + Unpin + 'static, + { while let Some(packet) = l3_packet_stream.next().await { let mut packet = match packet { Err(e) => { @@ -125,37 +131,61 @@ where header[0] = USER_DATA_VERSION; header[1] = USER_DATA_L3_TYPE; - // Get shared secret from node and dest address - let shared_secret = match router.get_shared_secret_from_dest(dst_ip) { - Some(ss) => ss, - None => { - debug!( - "No entry found for destination address {}, dropping packet", - dst_ip - ); - continue; - } - }; - - router.route_packet(DataPacket { - dst_ip, - src_ip, - raw_data: shared_secret.encrypt(packet), - }); + self.encrypt_and_route_packet(src_ip, dst_ip, packet) } warn!("Data inject loop from host to router ended"); } - async fn extract_l3_packet_loop( - router: Router, - mut l3_packet_sink: T, - mut host_packet_source: UnboundedReceiver, + /// Inject a new packet where the content is a `message` fragment. + pub fn inject_message_packet( + &self, + src_ip: Ipv6Addr, + dst_ip: Ipv6Addr, + mut packet: PacketBuffer, ) { + let mut header = packet.header_mut(); + header[0] = USER_DATA_VERSION; + header[1] = USER_DATA_MESSAGE_TYPE; + + self.encrypt_and_route_packet(src_ip, dst_ip, packet) + } + + fn encrypt_and_route_packet(&self, src_ip: Ipv6Addr, dst_ip: Ipv6Addr, packet: PacketBuffer) { + // Get shared secret from node and dest address + let shared_secret = match self.router.get_shared_secret_from_dest(dst_ip) { + Some(ss) => ss, + None => { + debug!( + "No entry found for destination address {}, dropping packet", + dst_ip + ); + return; + } + }; + + self.router.route_packet(DataPacket { + dst_ip, + src_ip, + raw_data: shared_secret.encrypt(packet), + }); + } + + async fn extract_packet_loop( + self, + mut l3_packet_sink: T, + mut message_packet_sink: U, + mut host_packet_source: UnboundedReceiver, + ) where + T: Sink + Send + Unpin + 'static, + T::Error: std::fmt::Display, + U: Sink + Send + Unpin + 'static, + U::Error: std::fmt::Display, + { while let Some(data_packet) = host_packet_source.recv().await { // decrypt & send to TUN interface let shared_secret = - if let Some(ss) = router.get_shared_secret_from_dest(data_packet.src_ip) { + if let Some(ss) = self.router.get_shared_secret_from_dest(data_packet.src_ip) { ss } else { trace!("Received packet from unknown sender"); @@ -177,14 +207,23 @@ where } // Route based on packet type. - if header[1] != USER_DATA_L3_TYPE { - trace!("Dropping decrypted packet with unknown protocol type"); - continue; - } - - if let Err(e) = l3_packet_sink.send(decrypted_packet).await { - error!("Failed to send packet on local TUN interface: {e}",); - continue; + match header[1] { + USER_DATA_L3_TYPE => { + if let Err(e) = l3_packet_sink.send(decrypted_packet).await { + error!("Failed to send packet on local TUN interface: {e}",); + continue; + } + } + USER_DATA_MESSAGE_TYPE => { + if let Err(e) = message_packet_sink.send(decrypted_packet).await { + error!("Failed to send packet to message handler: {e}",); + continue; + } + } + _ => { + trace!("Dropping decrypted packet with unknown protocol type"); + continue; + } } } diff --git a/src/lib.rs b/src/lib.rs index f7a30bc..b5d1c04 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,6 +4,7 @@ pub mod data; pub mod filters; mod interval; mod ip_pubkey; +pub mod message; mod metric; pub mod packet; mod peer; diff --git a/src/main.rs b/src/main.rs index b38d079..e192495 100644 --- a/src/main.rs +++ b/src/main.rs @@ -157,7 +157,14 @@ async fn main() -> Result<(), Box> { let _peer_manager: peer_manager::PeerManager = peer_manager::PeerManager::new(router.clone(), static_peers, cli.port); - let _data_plane = DataPlane::new(router.clone(), rxhalf, txhalf, tun_rx); + let _data_plane = DataPlane::new( + router.clone(), + rxhalf, + txhalf, + // TODO: proper sink to message handler + futures::sink::drain(), + tun_rx, + ); // TODO: put in dedicated file so we can only rely on certain signals on unix platforms let mut sigusr1 = diff --git a/src/message.rs b/src/message.rs new file mode 100644 index 0000000..22f269a --- /dev/null +++ b/src/message.rs @@ -0,0 +1,206 @@ +//! Module for working with "messages". +//! +//! A message is an arbitrary bag of bytes sent by a node to a different node. A message is +//! considered application defined data (L7), and we make no assumptions of any kind regarding the +//! structure. We only care about sending the message to the remote in the most reliable way +//! possible. + +use std::{ + collections::HashMap, + net::Ipv6Addr, + sync::{Arc, Mutex}, + time::{self, Duration}, +}; + +use rand::Fill; + +use crate::data::DataPlane; + +/// The size in bytes of the message header which starts each user message packet. +const MESSAGE_HEADER_SIZE: usize = 12; + +/// Flag indicating we are starting a new message. The message ID is specified in the header. The +/// body contains the length of the message. The receiver must create an entry for the new ID. This +/// flag must always be set on the first packet of a message stream. If a receiver already received +/// data for this message and a new packet comes in with this flag set for this message, all +/// existing data must be removed on the receiver side. +const FLAG_MESSAGE_INIT: u8 = 0b1000_0000; +// Flag indicating the message with the given ID is done, i.e. it has been fully transmitted. +const FLAG_MESSAGE_DONE: u8 = 0b0100_0000; +/// Indicats the message with this ID is aborted by the sender and the receiver should discard it. +/// The receiver can ignore this if it fully received the message. +const FLAG_MESSAGE_ABORTED: u8 = 0b0010_0000; +/// Flag indicating the message with the given ID has been read by the receiver, that is it has +/// been transfered to an external process. +const FLAG_MESSAGE_READ: u8 = 0b0001_0000; +/// Flag acknowledging receipt of a chunk. Once this has been received, the packet __should not__ be +/// transmitted again by the sender. +const FLAG_MESSAGE_CHUNK_ACK: u8 = 0b0000_0001; + +pub struct MessageStack { + data_plane: Arc, + inbox: Arc>, + outbox: Arc>, +} + +struct MessageOutbox { + msges: HashMap, +} + +struct MessageInbox { + /// Messages which are still being transmitted. + pending_msges: HashMap, + /// Messages which have been completed. + complete_msges: HashMap, +} + +struct ReceivedMessageInfo { + id: MessageId, + src: Ipv6Addr, + dst: Ipv6Addr, + message: Vec, +} + +/// A chunk of a message. This represents individual data pieces on the receiver side. +struct Chunk { + chunk_idx: usize, + data: Vec, +} + +/// Description of an individual chunk. +struct ChunkState { + /// Index of the chunk in the chunk stream. + chunk_idx: usize, + /// Offset of the chunk in the message. + chunk_offset: usize, + /// Size of the chunk. + // TODO: is this needed or can this be extrapolated by checking the next chunk in the list? + chunk_size: usize, + /// Transmit state of the chunk. + chunk_transmit_state: ChunkTransmitState, +} + +/// Transmission state of an individual chunk +enum ChunkTransmitState { + /// The chunk hasn't been transmitted yet. + Started, + /// The chunk has been sent but we did not receive an acknowledgment yet. The time the chunk + /// was sent is remembered so we can calulcate if we need to try sending it again. + Sent(std::time::Instant), + /// The receiver has acknowledged receipt of the chunk. + Acked, +} + +enum TransmissionState { + /// Transmission has not started yet. + Init, + /// Transmission is in progress. + InProgress(Vec), + /// Remote acknowledged full reception. + Received, + /// Remote indicated the message has been read by an external entity. + Read, + /// Transmission aborted by us. We indicated this by sending an abort flag to the receiver. + Aborted, +} + +impl MessageInbox { + fn new() -> Self { + Self { + pending_msges: HashMap::new(), + complete_msges: HashMap::new(), + } + } +} + +impl MessageOutbox { + /// Create a new `MessageOutbox` ready for use. + fn new() -> Self { + Self { + msges: HashMap::new(), + } + } + + /// Insert a new message for tracking during (and after) sending. + fn insert(&mut self, msg: OutboundMessageInfo) { + self.msges.insert(msg.msg.id, msg); + } +} + +impl MessageStack { + pub fn new(data_plane: DataPlane) -> Arc { + Arc::new(Self { + data_plane: Arc::new(data_plane), + inbox: Arc::new(Mutex::new(MessageInbox::new())), + outbox: Arc::new(Mutex::new(MessageOutbox::new())), + }) + } +} + +pub struct MessagePacketHeader { + data: [u8; MESSAGE_HEADER_SIZE], +} + +impl MessageStack { + /// Push a new message to be transmitted, which will be tried for the given duration. + pub fn push_message(&self, dst: Ipv6Addr, data: Vec, try_duration: Duration) -> MessageId { + let src = self.data_plane.router().node_public_key().address(); + + let id = MessageId::new(); + + let msg = Message { id, src, dst, data }; + + let created = std::time::SystemTime::now(); + let deadline = created + try_duration; + + let obmi = OutboundMessageInfo { + state: TransmissionState::Init, + created, + deadline, + msg, + }; + + self.outbox + .lock() + .expect("Outbox lock isn't poisoned; qed") + .insert(obmi); + + id + } +} + +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct MessageId([u8; 8]); + +impl MessageId { + fn new() -> Self { + let mut id = Self([0u8; 8]); + + id.0.try_fill(&mut rand::thread_rng()) + .expect("Can instantiate new ID from thread RNG generator; qed"); + + id + } +} + +pub struct Message { + /// Generated ID, used to identify the message on the wire + id: MessageId, + /// Source IP (ours) + src: Ipv6Addr, + /// Destination IP + dst: Ipv6Addr, + /// Data + data: Vec, +} + +pub struct OutboundMessageInfo { + /// The current state of the message + state: TransmissionState, + /// Timestamp when the message was created (received by this node). + created: time::SystemTime, + /// Timestamp indicating when we stop trying to send the message. + deadline: time::SystemTime, + /// The message to send + msg: Message, +}