Initial message structure

Signed-off-by: Lee Smet <lee.smet@hotmail.com>

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<S, T> {
+pub struct DataPlane {
     router: Router,
-    _stream_marker: PhantomData<S>,
-    _sink_marker: PhantomData<T>,
 }
 
-impl<S, T> DataPlane<S, T>
-where
-    S: Stream<Item = Result<PacketBuffer, std::io::Error>> + Send + Unpin + 'static,
-    T: Sink<PacketBuffer> + 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<S, T, U>(
         router: Router,
         l3_packet_stream: S,
         l3_packet_sink: T,
+        message_packet_sink: U,
         host_packet_source: UnboundedReceiver<DataPacket>,
-    ) -> 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<Item = Result<PacketBuffer, std::io::Error>> + Send + Unpin + 'static,
+        T: Sink<PacketBuffer> + Send + Unpin + 'static,
+        T::Error: std::fmt::Display,
+        U: Sink<PacketBuffer> + 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<S>(self, mut l3_packet_stream: S)
+    where
+        // TODO: no result
+        // TODO: should IP extraction be handled higher up?
+        S: Stream<Item = Result<PacketBuffer, std::io::Error>> + 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<DataPacket>,
+    /// 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<T, U>(
+        self,
+        mut l3_packet_sink: T,
+        mut message_packet_sink: U,
+        mut host_packet_source: UnboundedReceiver<DataPacket>,
+    ) where
+        T: Sink<PacketBuffer> + Send + Unpin + 'static,
+        T::Error: std::fmt::Display,
+        U: Sink<PacketBuffer> + 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;
+                }
             }
         }
 

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;

src/main.rs

@@ -157,7 +157,14 @@ async fn main() -> Result<(), Box<dyn Error>> {
     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 =

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<DataPlane>,
+    inbox: Arc<Mutex<MessageInbox>>,
+    outbox: Arc<Mutex<MessageOutbox>>,
+}
+
+struct MessageOutbox {
+    msges: HashMap<MessageId, OutboundMessageInfo>,
+}
+
+struct MessageInbox {
+    /// Messages which are still being transmitted.
+    pending_msges: HashMap<MessageId, ReceivedMessageInfo>,
+    /// Messages which have been completed.
+    complete_msges: HashMap<MessageId, Message>,
+}
+
+struct ReceivedMessageInfo {
+    id: MessageId,
+    src: Ipv6Addr,
+    dst: Ipv6Addr,
+    message: Vec<Chunk>,
+}
+
+/// A chunk of a message. This represents individual data pieces on the receiver side.
+struct Chunk {
+    chunk_idx: usize,
+    data: Vec<u8>,
+}
+
+/// 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<ChunkState>),
+    /// 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<Self> {
+        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<u8>, 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<u8>,
+}
+
+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,
+}