Merge branch 'release/0.1.0'

Cargo.toml

@@ -7,5 +7,10 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+base64 = "0.11.0"
+clap = "2.33.0"
+curve25519-dalek = "1.2.3"
+futures = "0.3.1"
+nym-client = { path = "../nym-client" }
 sphinx = { path = "../sphinx" }
-tokio = { version = "0.2", features = ["full"] }
+tokio = { version = "0.2", features = ["full"] }

README.md

@@ -0,0 +1,19 @@
+# Nym Mixnode
+
+A Rust mixnode implementation.
+
+## Building
+
+* check out the code
+* [install rust](https://www.rust-lang.org/tools/install) (stable)
+* `cargo build --release` (for a production build)
+
+The built binary can be found at `target/release/nym-mixnode`
+
+## Usage
+
+* `nym-mixnode` prints a help message showing usage options
+* `nym-mixnode run --help` prints a help message showing usage options for the run command
+* `nym-mixnode run --layer 1` will start the mixnode in layer 1 (coordinate with other people to find out which layer you need to start yours in)
+
+By default, the Nym Mixnode will start on port 1789. If desired, you can change the port using the `--port` option.

src/main.rs

@@ -1,51 +1,73 @@
-use tokio::net::TcpListener;
-use tokio::prelude::*;
-use crate::mix_peer::MixPeer;
-use sphinx::SphinxPacket;
+use clap::{App, Arg, ArgMatches, SubCommand};
+use std::process;
 
 mod mix_peer;
+mod node;
 
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let my_address = "127.0.0.1:8080";
-    let mut listener = TcpListener::bind(my_address).await?;
+fn main() {
+    let arg_matches = App::new("Nym Mixnode")
+        .version("0.1.0")
+        .author("Nymtech")
+        .about("Implementation of the Loopix-based Mixnode")
+        .subcommand(
+            SubCommand::with_name("run")
+                .about("Starts the mixnode")
+                .arg(
+                    Arg::with_name("host")
+                        .long("host")
+                        .help("The custom host on which the mixnode will be running")
+                        .takes_value(true),
+                )
+                .arg(
+                    Arg::with_name("port")
+                        .long("port")
+                        .help("The port on which the mixnode will be listening")
+                        .takes_value(true)
+                )
+                .arg(
+                    Arg::with_name("layer")
+                        .long("layer")
+                        .help("The mixnet layer of this particular node")
+                        .takes_value(true)
+                        .required(true),
+                )
+                .arg(
+                    Arg::with_name("local")
+                        .long("local")
+                        .help("Flag to indicate whether the client is expected to run on a local deployment.")
+                        .takes_value(false),
+                ),
+        )
+        .get_matches();
 
-    println!("Starting Nym mixnode on address {:?}", my_address);
-    println!("Waiting for input...");
-
-    loop {
-        let (mut inbound, _) = listener.accept().await?;
-
-        tokio::spawn(async move {
-            let mut buf = [0; 1024];
-
-            loop {
-                let _ = match inbound.read(&mut buf).await {
-                    Ok(length) if length == 0 =>
-                        {
-                            println!("Remote connection closed.");
-                            return
-                        }
-                    Ok(length) => {
-                        let packet = SphinxPacket::from_bytes(buf.to_vec()).unwrap();
-                        let (next_packet, _) = packet.process(Default::default());
-
-                        let next_mix = MixPeer::new();
-
-                        match next_mix.send(next_packet.to_bytes()).await {
-                            Ok(()) => length,
-                            Err(e) => {
-                                println!("failed to write bytes to next mix peer. err = {:?}", e.to_string());
-                                return;
-                            }
-                        }
-                    }
-                    Err(e) => {
-                        println!("failed to read from socket; err = {:?}", e);
-                        return;
-                    }
-                };
-            }
-        });
+    if let Err(e) = execute(arg_matches) {
+        println!("{}", e);
+        process::exit(1);
     }
 }
+
+fn execute(matches: ArgMatches) -> Result<(), String> {
+    match matches.subcommand() {
+        ("run", Some(m)) => Ok(node::runner::start(m)),
+        _ => Err(usage()),
+    }
+}
+
+fn usage() -> String {
+    banner() + "usage: --help to see available options.\n\n"
+}
+
+fn banner() -> String {
+    return r#"
+
+      _ __  _   _ _ __ ___
+     | '_ \| | | | '_ \ _ \
+     | | | | |_| | | | | | |
+     |_| |_|\__, |_| |_| |_|
+            |___/
+
+             (mixnode)
+
+    "#
+    .to_string();
+}

src/mix_peer.rs

@@ -1,24 +1,32 @@
-use tokio::net::TcpStream;
-use tokio::prelude::*;
 use std::error::Error;
+use std::net::{Ipv4Addr, SocketAddrV4};
+use tokio::prelude::*;
 
-
+#[derive(Debug)]
 pub struct MixPeer {
-    connection: &'static str,
+    connection: SocketAddrV4,
 }
 
 impl MixPeer {
-    pub fn new() -> MixPeer {
-        let next_hop_address = "127.0.0.1:8081";
-        let node = MixPeer {
-            connection: next_hop_address,
-        };
-        node
+    // note that very soon `next_hop_address` will be changed to `next_hop_metadata`
+    pub fn new(next_hop_address: [u8; 32]) -> MixPeer {
+        let b = next_hop_address;
+        let host = Ipv4Addr::new(b[0], b[1], b[2], b[3]);
+        let port: u16 = u16::from_be_bytes([b[4], b[5]]);
+        let socket_address = SocketAddrV4::new(host, port);
+        MixPeer {
+            connection: socket_address,
+        }
     }
 
-    pub async fn send(&self, bytes: Vec<u8>) -> Result<(), Box<dyn Error>>{
-        let mut stream = TcpStream::connect(self.connection).await?;
+    pub async fn send(&self, bytes: Vec<u8>) -> Result<(), Box<dyn Error>> {
+        let next_hop_address = self.connection.clone();
+        let mut stream = tokio::net::TcpStream::connect(next_hop_address).await?;
         stream.write_all(&bytes).await?;
         Ok(())
     }
-}
+
+    pub fn to_string(&self) -> String {
+        self.connection.to_string()
+    }
+}

src/node/metrics.rs

@@ -0,0 +1,94 @@
+use futures::channel::mpsc;
+use futures::lock::Mutex;
+use futures::StreamExt;
+use nym_client::clients::directory;
+use nym_client::clients::directory::metrics::MixMetric;
+use nym_client::clients::directory::requests::metrics_mixes_post::MetricsMixPoster;
+use nym_client::clients::directory::DirectoryClient;
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+const METRICS_INTERVAL: u64 = 3;
+
+#[derive(Debug)]
+pub struct MetricsReporter {
+    received: u64,
+    sent: HashMap<String, u64>,
+}
+
+impl MetricsReporter {
+    pub(crate) fn new() -> Self {
+        MetricsReporter {
+            received: 0,
+            sent: HashMap::new(),
+        }
+    }
+
+    pub(crate) fn add_arc_mutex(self) -> Arc<Mutex<Self>> {
+        Arc::new(Mutex::new(self))
+    }
+
+    async fn increment_received_metrics(metrics: Arc<Mutex<MetricsReporter>>) {
+        let mut unlocked = metrics.lock().await;
+        unlocked.received += 1;
+    }
+
+    pub(crate) async fn run_received_metrics_control(
+        metrics: Arc<Mutex<MetricsReporter>>,
+        mut rx: mpsc::Receiver<()>,
+    ) {
+        while let Some(_) = rx.next().await {
+            MetricsReporter::increment_received_metrics(metrics.clone()).await;
+        }
+    }
+
+    async fn increment_sent_metrics(metrics: Arc<Mutex<MetricsReporter>>, sent_to: String) {
+        let mut unlocked = metrics.lock().await;
+        let receiver_count = unlocked.sent.entry(sent_to).or_insert(0);
+        *receiver_count += 1;
+    }
+
+    pub(crate) async fn run_sent_metrics_control(
+        metrics: Arc<Mutex<MetricsReporter>>,
+        mut rx: mpsc::Receiver<String>,
+    ) {
+        while let Some(sent_metric) = rx.next().await {
+            MetricsReporter::increment_sent_metrics(metrics.clone(), sent_metric).await;
+        }
+    }
+
+    async fn acquire_and_reset_metrics(
+        metrics: Arc<Mutex<MetricsReporter>>,
+    ) -> (u64, HashMap<String, u64>) {
+        let mut unlocked = metrics.lock().await;
+        let received = unlocked.received;
+
+        let sent = std::mem::replace(&mut unlocked.sent, HashMap::new());
+        unlocked.received = 0;
+
+        (received, sent)
+    }
+
+    pub(crate) async fn run_metrics_sender(
+        metrics: Arc<Mutex<MetricsReporter>>,
+        cfg: directory::Config,
+        pub_key_str: String,
+    ) {
+        let delay_duration = Duration::from_secs(METRICS_INTERVAL);
+        let directory_client = directory::Client::new(cfg);
+        loop {
+            tokio::time::delay_for(delay_duration).await;
+            let (received, sent) =
+                MetricsReporter::acquire_and_reset_metrics(metrics.clone()).await;
+            directory_client
+                .metrics_post
+                .post(&MixMetric {
+                    pub_key: pub_key_str.clone(),
+                    received,
+                    sent,
+                })
+                .unwrap();
+        }
+    }
+}

src/node/mod.rs

@@ -0,0 +1,263 @@
+use curve25519_dalek::montgomery::MontgomeryPoint;
+use curve25519_dalek::scalar::Scalar;
+use futures::channel::mpsc;
+use futures::lock::Mutex;
+use sphinx::header::delays::Delay as SphinxDelay;
+use sphinx::{ProcessedPacket, SphinxPacket};
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::prelude::*;
+use tokio::runtime::Runtime;
+
+use crate::mix_peer::MixPeer;
+use crate::node::metrics::MetricsReporter;
+use futures::SinkExt;
+use nym_client::clients::directory;
+
+mod metrics;
+mod presence;
+pub mod runner;
+
+pub struct Config {
+    directory_server: String,
+    layer: usize,
+    public_key: MontgomeryPoint,
+    secret_key: Scalar,
+    socket_address: SocketAddr,
+}
+
+impl Config {
+    pub fn public_key_string(&self) -> String {
+        let key_bytes = self.public_key.to_bytes().to_vec();
+        base64::encode_config(&key_bytes, base64::URL_SAFE)
+    }
+}
+
+#[derive(Debug)]
+pub enum MixProcessingError {
+    SphinxRecoveryError,
+    ReceivedFinalHopError,
+}
+
+impl From<sphinx::ProcessingError> for MixProcessingError {
+    // for time being just have a single error instance for all possible results of sphinx::ProcessingError
+    fn from(_: sphinx::ProcessingError) -> Self {
+        use MixProcessingError::*;
+
+        SphinxRecoveryError
+    }
+}
+
+struct ForwardingData {
+    packet: SphinxPacket,
+    delay: SphinxDelay,
+    recipient: MixPeer,
+    sent_metrics_tx: mpsc::Sender<String>,
+}
+
+// TODO: this will need to be changed if MixPeer will live longer than our Forwarding Data
+impl ForwardingData {
+    fn new(
+        packet: SphinxPacket,
+        delay: SphinxDelay,
+        recipient: MixPeer,
+        sent_metrics_tx: mpsc::Sender<String>,
+    ) -> Self {
+        ForwardingData {
+            packet,
+            delay,
+            recipient,
+            sent_metrics_tx,
+        }
+    }
+}
+
+// ProcessingData defines all data required to correctly unwrap sphinx packets
+struct ProcessingData {
+    secret_key: Scalar,
+    received_metrics_tx: mpsc::Sender<()>,
+    sent_metrics_tx: mpsc::Sender<String>,
+}
+
+impl ProcessingData {
+    fn new(
+        secret_key: Scalar,
+        received_metrics_tx: mpsc::Sender<()>,
+        sent_metrics_tx: mpsc::Sender<String>,
+    ) -> Self {
+        ProcessingData {
+            secret_key,
+            received_metrics_tx,
+            sent_metrics_tx,
+        }
+    }
+
+    fn add_arc_mutex(self) -> Arc<Mutex<Self>> {
+        Arc::new(Mutex::new(self))
+    }
+}
+
+struct PacketProcessor {}
+
+impl PacketProcessor {
+    pub async fn process_sphinx_data_packet(
+        packet_data: &[u8],
+        processing_data: Arc<Mutex<ProcessingData>>,
+    ) -> Result<ForwardingData, MixProcessingError> {
+        // we received something resembling a sphinx packet, report it!
+        let processing_data = processing_data.lock().await;
+        let mut received_sender = processing_data.received_metrics_tx.clone();
+
+        received_sender.send(()).await.unwrap();
+
+        let packet = SphinxPacket::from_bytes(packet_data.to_vec())?;
+        let (next_packet, next_hop_address, delay) =
+            match packet.process(processing_data.secret_key) {
+                ProcessedPacket::ProcessedPacketForwardHop(packet, address, delay) => {
+                    (packet, address, delay)
+                }
+                _ => return Err(MixProcessingError::ReceivedFinalHopError),
+            };
+
+        let next_mix = MixPeer::new(next_hop_address);
+
+        let fwd_data = ForwardingData::new(
+            next_packet,
+            delay,
+            next_mix,
+            processing_data.sent_metrics_tx.clone(),
+        );
+        Ok(fwd_data)
+    }
+
+    async fn wait_and_forward(mut forwarding_data: ForwardingData) {
+        let delay_duration = Duration::from_nanos(forwarding_data.delay.get_value());
+        tokio::time::delay_for(delay_duration).await;
+        forwarding_data
+            .sent_metrics_tx
+            .send(forwarding_data.recipient.to_string())
+            .await
+            .unwrap();
+
+        println!("RECIPIENT: {:?}", forwarding_data.recipient);
+        match forwarding_data
+            .recipient
+            .send(forwarding_data.packet.to_bytes())
+            .await
+        {
+            Ok(()) => (),
+            Err(e) => {
+                println!(
+                    "failed to write bytes to next mix peer. err = {:?}",
+                    e.to_string()
+                );
+            }
+        }
+    }
+}
+
+// the MixNode will live for whole duration of this program
+pub struct MixNode {
+    directory_server: String,
+    network_address: SocketAddr,
+    public_key: MontgomeryPoint,
+    secret_key: Scalar,
+    // TODO: use it later to enforce forward travel
+    //    layer: usize,
+}
+
+impl MixNode {
+    pub fn new(config: &Config) -> Self {
+        MixNode {
+            directory_server: config.directory_server.clone(),
+            network_address: config.socket_address,
+            secret_key: config.secret_key,
+            public_key: config.public_key,
+            //            layer: config.layer,
+        }
+    }
+
+    async fn process_socket_connection(
+        mut socket: tokio::net::TcpStream,
+        processing_data: Arc<Mutex<ProcessingData>>,
+    ) {
+        // NOTE: processing_data is copied here!!
+        let mut buf = [0u8; sphinx::PACKET_SIZE];
+
+        // In a loop, read data from the socket and write the data back.
+        loop {
+            match socket.read(&mut buf).await {
+                // socket closed
+                Ok(n) if n == 0 => {
+                    println!("Remote connection closed.");
+                    return;
+                }
+                Ok(_) => {
+                    let fwd_data = PacketProcessor::process_sphinx_data_packet(
+                        buf.as_ref(),
+                        processing_data.clone(),
+                    )
+                    .await
+                    .unwrap();
+                    PacketProcessor::wait_and_forward(fwd_data).await;
+                }
+                Err(e) => {
+                    println!("failed to read from socket; err = {:?}", e);
+                    return;
+                }
+            };
+
+            // Write the some data back
+            if let Err(e) = socket.write_all(b"foomp").await {
+                println!("failed to write reply to socket; err = {:?}", e);
+                return;
+            }
+        }
+    }
+
+    pub fn start(&self) -> Result<(), Box<dyn std::error::Error>> {
+        // Create the runtime, probably later move it to MixNode itself?
+        let mut rt = Runtime::new()?;
+
+        let (received_tx, received_rx) = mpsc::channel(1024);
+        let (sent_tx, sent_rx) = mpsc::channel(1024);
+
+        let directory_cfg = directory::Config {
+            base_url: self.directory_server.clone(),
+        };
+        let pub_key_str =
+            base64::encode_config(&self.public_key.to_bytes().to_vec(), base64::URL_SAFE);
+
+        let metrics = MetricsReporter::new().add_arc_mutex();
+        rt.spawn(MetricsReporter::run_received_metrics_control(
+            metrics.clone(),
+            received_rx,
+        ));
+        rt.spawn(MetricsReporter::run_sent_metrics_control(
+            metrics.clone(),
+            sent_rx,
+        ));
+        rt.spawn(MetricsReporter::run_metrics_sender(
+            metrics,
+            directory_cfg,
+            pub_key_str,
+        ));
+
+        // Spawn the root task
+        rt.block_on(async {
+            let mut listener = tokio::net::TcpListener::bind(self.network_address).await?;
+            let processing_data =
+                ProcessingData::new(self.secret_key, received_tx, sent_tx).add_arc_mutex();
+
+            loop {
+                let (socket, _) = listener.accept().await?;
+
+                let thread_processing_data = processing_data.clone();
+                tokio::spawn(async move {
+                    MixNode::process_socket_connection(socket, thread_processing_data).await;
+                });
+            }
+        })
+    }
+}

src/node/presence.rs

@@ -0,0 +1,45 @@
+use crate::node;
+use nym_client::clients::directory;
+use nym_client::clients::directory::presence::MixNodePresence;
+use nym_client::clients::directory::requests::presence_mixnodes_post::PresenceMixNodesPoster;
+use nym_client::clients::directory::DirectoryClient;
+use std::thread;
+use std::time::Duration;
+
+pub struct Notifier {
+    pub net_client: directory::Client,
+    presence: MixNodePresence,
+}
+
+impl Notifier {
+    pub fn new(node_config: &node::Config) -> Notifier {
+        let config = directory::Config {
+            base_url: node_config.directory_server.clone(),
+        };
+        let net_client = directory::Client::new(config);
+        let presence = MixNodePresence {
+            host: node_config.socket_address.to_string(), // note: the directory server determines the real incoming IP itself, but uses the socket. Host here is just a placeholder.
+            pub_key: node_config.public_key_string(),
+            layer: node_config.layer as u64,
+            last_seen: 0,
+        };
+        Notifier {
+            net_client,
+            presence,
+        }
+    }
+
+    pub fn notify(&self) {
+        self.net_client
+            .presence_mix_nodes_post
+            .post(&self.presence)
+            .unwrap();
+    }
+
+    pub fn run(&self) {
+        loop {
+            self.notify();
+            thread::sleep(Duration::from_secs(5));
+        }
+    }
+}

src/node/runner.rs

@@ -0,0 +1,63 @@
+use crate::banner;
+use crate::node;
+use crate::node::presence;
+use crate::node::MixNode;
+use clap::ArgMatches;
+
+use std::net::ToSocketAddrs;
+use std::thread;
+
+pub fn start(matches: &ArgMatches) {
+    println!("{}", banner());
+    println!("Starting mixnode...");
+
+    let config = new_config(matches);
+
+    // println!("Startup on: {}", config.socket_address);
+    println!("Listening for incoming packets...");
+
+    let mix = MixNode::new(&config);
+    thread::spawn(move || {
+        let notifier = presence::Notifier::new(&config);
+        notifier.run();
+    });
+    mix.start().unwrap();
+}
+
+fn new_config(matches: &ArgMatches) -> node::Config {
+    let host = matches.value_of("host").unwrap_or("0.0.0.0");
+
+    let port = match matches.value_of("port").unwrap_or("1789").parse::<u16>() {
+        Ok(n) => n,
+        Err(err) => panic!("Invalid port value provided - {:?}", err),
+    };
+
+    let layer = match matches.value_of("layer").unwrap().parse::<usize>() {
+        Ok(n) => n,
+        Err(err) => panic!("Invalid layer value provided - {:?}", err),
+    };
+
+    let is_local = matches.is_present("local");
+
+    let socket_address = (host, port)
+        .to_socket_addrs()
+        .expect("Failed to combine host and port")
+        .next()
+        .expect("Failed to extract the socket address from the iterator");
+
+    let (secret_key, public_key) = sphinx::crypto::keygen();
+
+    let directory_server = if is_local {
+        "http://localhost:8080".to_string()
+    } else {
+        "https://directory.nymtech.net".to_string()
+    };
+
+    node::Config {
+        directory_server,
+        layer,
+        public_key,
+        socket_address,
+        secret_key,
+    }
+}