Merge branch 'release/v0.3.3'

2020-01-20 17:52:22 +00:00
parent fe88e20350 89087a3dfd
commit bec0223899
16 changed files with 271 additions and 94 deletions
@@ -7,3 +7,9 @@ jobs:
  allow_failures:
    - rust: nightly
  fast_finish: true
+before_script:
+  - rustup component add rustfmt
+script:
+  - cargo build
+  - cargo test
+  - cargo fmt -- --check
@@ -1339,7 +1339,7 @@ dependencies = [

 [[package]]
 name = "nym-client"
-version = "0.3.2"
+version = "0.3.3"
 dependencies = [
 "addressing",
 "base64 0.11.0",
@@ -1372,7 +1372,7 @@ dependencies = [

 [[package]]
 name = "nym-mixnode"
-version = "0.3.2"
+version = "0.3.3"
 dependencies = [
 "addressing",
 "base64 0.11.0",
@@ -1388,7 +1388,7 @@ dependencies = [

 [[package]]
 name = "nym-sfw-provider"
-version = "0.3.2"
+version = "0.3.3"
 dependencies = [
 "base64 0.11.0",
 "built",
@@ -2381,7 +2381,7 @@ dependencies = [
 [[package]]
 name = "tokio-tungstenite"
 version = "0.10.0"
-source = "git+https://github.com/dbcfd/tokio-tungstenite?rev=6dc2018cbfe8fe7ddd75ff977343086503135b38#6dc2018cbfe8fe7ddd75ff977343086503135b38"
+source = "git+https://github.com/snapview/tokio-tungstenite?rev=308d9680c0e59dd1e8651659a775c05df937934e#308d9680c0e59dd1e8651659a775c05df937934e"
 dependencies = [
 "futures 0.3.1",
 "log",
@@ -2,11 +2,12 @@

 This repository contains the full Nym platform, written in Rust.

-The platform is composed of multiple Rust crates. Top-level crates include:
+The platform is composed of multiple Rust crates. Top-level executable binary crates include:

-* client - an executable crate which you can use for interacting with Nym nodes
-* mixnode - an executable mixnode crate
-* sfw-provider - an executable store-and-forward provider crate. The provider acts sort of like a mailbox for mixnet messages.
+* client - an executable which you can build into your own applications. Use it for interacting with Nym nodes.
+* mixnode - the mixnode crate.
+* sfw-provider - a store-and-forward service provider. The provider acts sort of like a mailbox for mixnet messages.
+* validator - currently just starting development. Handles consensus ordering of transactions, mixmining, and coconut credential generation and validation. 

 [![Build Status](https://travis-ci.com/nymtech/nym.svg?branch=develop)](https://travis-ci.com/nymtech/nym)

@@ -1,5 +1,9 @@
 # nym-mixnode Changelog

+## 0.3.3
+
+* Version increase for consistency with `nym-client`
+
 ## 0.3.2

 * added separate announce address
@@ -1,7 +1,7 @@
 [package]
 build = "build.rs"
 name = "nym-mixnode"
-version = "0.3.2"
+version = "0.3.3"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
 edition = "2018"

@@ -1,5 +1,12 @@
 # nym-client Changelog

+## 0.3.3
+
+* websocket handling of 'ping', 'pong' and 'close' messages
+* websocket not crashing on binary messages
+* websocket returning text rather than base64
+* restored `nym-client` lib functionality 
+
 ## 0.3.2

 * allows receiving topology with dns hostname instead of an ip address
@@ -1,12 +1,16 @@
 [package]
 build = "build.rs"
 name = "nym-client"
-version = "0.3.2"
+version = "0.3.3"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
 edition = "2018"

 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

+[lib]
+name = "nym_client"
+path = "src/lib.rs"
+
 [dependencies]
 base64 = "0.11.0"
 clap = "2.33.0"
@@ -39,7 +43,7 @@ topology = {path = "../common/topology" }
 sphinx = { git = "https://github.com/nymtech/sphinx", rev="1d8cefcb6a0cb8e87d00d89eb1ccf2839e92aa1f" }

 # putting this explicitly below everything and most likely, the next time we look into it, it will already have a proper release
-tokio-tungstenite = { git = "https://github.com/dbcfd/tokio-tungstenite", rev="6dc2018cbfe8fe7ddd75ff977343086503135b38" }
+tokio-tungstenite = { git = "https://github.com/snapview/tokio-tungstenite", rev="308d9680c0e59dd1e8651659a775c05df937934e" }

 [build-dependencies]
 built = "0.3.2"
@@ -0,0 +1,2 @@
+// The file has been placed there by the build script.
+include!(concat!(env!("OUT_DIR"), "/built.rs"));
@@ -1,11 +1,9 @@
-use crate::banner;
 use crate::persistence::pathfinder::Pathfinder;
 use crate::persistence::pemstore::PemStore;
 use clap::ArgMatches;
 use crypto::identity::MixnetIdentityKeyPair;

 pub fn execute(matches: &ArgMatches) {
-    println!("{}", banner());
    println!("Initialising client...");

    let id = matches.value_of("id").unwrap().to_string(); // required for now
@@ -1,4 +1,3 @@
-use crate::banner;
 use crate::clients::{NymClient, SocketType};
 use crate::persistence::pemstore;

@@ -7,8 +6,6 @@ use crypto::identity::{MixnetIdentityKeyPair, MixnetIdentityPublicKey};
 use std::net::ToSocketAddrs;

 pub fn execute(matches: &ArgMatches) {
-    println!("{}", banner());
-
    let id = matches.value_of("id").unwrap().to_string();
    let port = match matches.value_of("port").unwrap_or("9001").parse::<u16>() {
        Ok(n) => n,
@@ -1,4 +1,3 @@
-use crate::banner;
 use crate::clients::{NymClient, SocketType};
 use crate::persistence::pemstore;

@@ -7,8 +6,6 @@ use crypto::identity::{MixnetIdentityKeyPair, MixnetIdentityPublicKey};
 use std::net::ToSocketAddrs;

 pub fn execute(matches: &ArgMatches) {
-    println!("{}", banner());
-
    let id = matches.value_of("id").unwrap().to_string();
    let port = match matches.value_of("port").unwrap_or("9001").parse::<u16>() {
        Ok(n) => n,
@@ -0,0 +1,8 @@
+#![recursion_limit = "256"]
+
+pub mod built_info;
+pub mod clients;
+pub mod commands;
+pub mod persistence;
+pub mod sockets;
+pub mod utils;
@@ -1,19 +1,13 @@
 #![recursion_limit = "256"]

 use clap::{App, Arg, ArgMatches, SubCommand};
-use env_logger;
-use log::*;
-use std::process;

+pub mod built_info;
 pub mod clients;
 mod commands;
 mod persistence;
 mod sockets;
 pub mod utils;
-pub mod built_info {
-    // The file has been placed there by the build script.
-    include!(concat!(env!("OUT_DIR"), "/built.rs"));
-}

 fn main() {
    env_logger::init();
@@ -86,18 +80,26 @@ fn main() {
        )
        .get_matches();

-    if let Err(e) = execute(arg_matches) {
-        error!("{}", e);
-        process::exit(1);
-    }
+    execute(arg_matches);
 }

-fn execute(matches: ArgMatches) -> Result<(), String> {
+fn execute(matches: ArgMatches) {
    match matches.subcommand() {
-        ("init", Some(m)) => Ok(commands::init::execute(m)),
-        ("tcpsocket", Some(m)) => Ok(commands::tcpsocket::execute(m)),
-        ("websocket", Some(m)) => Ok(commands::websocket::execute(m)),
-        _ => Err(usage()),
+        ("init", Some(m)) => {
+            println!("{}", banner());
+            commands::init::execute(m);
+        }
+        ("tcpsocket", Some(m)) => {
+            println!("{}", banner());
+            commands::tcpsocket::execute(m);
+        }
+        ("websocket", Some(m)) => {
+            println!("{}", banner());
+            commands::websocket::execute(m);
+        }
+        _ => {
+            println!("{}", usage());
+        }
    }
 }

@@ -6,12 +6,14 @@ use futures::channel::{mpsc, oneshot};
 use futures::future::FutureExt;
 use futures::io::Error;
 use futures::{SinkExt, StreamExt};
-use log::*;
+use log::{debug, error, info, trace, warn};
 use serde::{Deserialize, Serialize};
 use sphinx::route::{Destination, DestinationAddressBytes};
+use std::convert::TryFrom;
 use std::io;
 use std::net::SocketAddr;
-use tungstenite::protocol::Message;
+use tungstenite::protocol::frame::coding::CloseCode;
+use tungstenite::protocol::{CloseFrame, Message};

 struct Connection {
    address: SocketAddr,
@@ -23,6 +25,122 @@ struct Connection {
    tx: UnboundedSender<Message>,
 }

+impl Connection {
+    async fn handle_text_message(&self, msg: String) -> ServerResponse {
+        debug!("Handling text message request");
+        trace!("Content: {:?}", msg.clone());
+
+        let request = match ClientRequest::try_from(msg) {
+            Ok(req) => req,
+            Err(err) => {
+                return ServerResponse::Error {
+                    // we failed to parse the request
+                    message: format!("received invalid request. err: {:?}", err),
+                };
+            }
+        };
+
+        match request {
+            ClientRequest::Send {
+                message,
+                recipient_address,
+            } => {
+                ClientRequest::handle_send(message, recipient_address, self.msg_input.clone()).await
+            }
+            ClientRequest::Fetch => ClientRequest::handle_fetch(self.msg_query.clone()).await,
+            ClientRequest::GetClients => {
+                ClientRequest::handle_get_clients(self.topology.clone()).await
+            }
+            ClientRequest::OwnDetails => ClientRequest::handle_own_details(self.self_address).await,
+        }
+    }
+
+    // Currently our websocket cannot handle binary data, so just close the connection
+    // with unsupported close code.
+    async fn handle_binary_message(&self, _msg: Vec<u8>) -> Message {
+        debug!("Handling binary message request");
+
+        Message::Close(Some(CloseFrame {
+            code: CloseCode::Unsupported,
+            reason: "binary messages aren't yet supported".into(),
+        }))
+    }
+
+    // As per RFC6455 5.5.2. and 5.5.3.:
+    // Upon receipt of a Ping frame, an endpoint MUST send a Pong frame in response.
+    // A Pong frame sent in response to a Ping frame must have identical
+    // "Application data" as found in the message body of the Ping frame
+    // being replied to.
+    async fn handle_ping_message(&self, msg: Vec<u8>) -> Message {
+        debug!("Handling binary ping request");
+
+        // As per RFC6455 5.5:
+        // All control frames MUST have a payload length of 125 bytes or less
+        if msg.len() > 125 {
+            return Message::Close(Some(CloseFrame {
+                code: CloseCode::Protocol,
+                reason: format!("ping message of length {} sent", msg.len()).into(),
+            }));
+        }
+
+        Message::Pong(msg)
+    }
+
+    // As per RFC6455 5.5.3.:
+    // A Pong frame MAY be sent unsolicited.  This serves as a
+    // unidirectional heartbeat.  A response to an unsolicited Pong frame is
+    // not expected.
+    // Realistically this handler should never be used,
+    // but since we're nice we will reply with a Pong containing original content
+    async fn handle_pong_message(&self, msg: Vec<u8>) -> Message {
+        debug!("Handling pong message request");
+
+        // As per RFC6455 5.5:
+        // All control frames MUST have a payload length of 125 bytes or less
+        if msg.len() > 125 {
+            return Message::Close(Some(CloseFrame {
+                code: CloseCode::Protocol,
+                reason: format!("ping message of length {} sent", msg.len()).into(),
+            }));
+        }
+
+        Message::Pong(msg)
+    }
+
+    // As per RFC6455 5.5.1.:
+    // If an endpoint receives a Close frame and did not previously send a
+    // Close frame, the endpoint MUST send a Close frame in response. (When
+    // sending a Close frame in response, the endpoint typically echos the
+    // status code it received.)
+    async fn handle_close_message(&self, close_frame: Option<CloseFrame<'static>>) -> Message {
+        debug!("Handling close message request");
+
+        Message::Close(close_frame)
+    }
+
+    async fn handle(mut self) {
+        while let Some(msg) = self.rx.next().await {
+            trace!("Received a message from {}: {}", self.address, msg);
+            let response_message = match msg {
+                Message::Text(text_message) => self.handle_text_message(text_message).await.into(),
+                Message::Binary(binary_message) => self.handle_binary_message(binary_message).await,
+                Message::Ping(ping_message) => self.handle_ping_message(ping_message).await,
+                Message::Pong(pong_message) => self.handle_pong_message(pong_message).await,
+                Message::Close(close_frame) => self.handle_close_message(close_frame).await,
+            };
+
+            if let Err(err) = self.tx.unbounded_send(response_message) {
+                error!(
+                    "Failed to send response message to accepted connections handler: {}.\n\
+                     Shutting off the connection handler",
+                    err
+                );
+                return;
+            }
+        }
+    }
+}
+
 #[derive(Debug)]
 pub enum WebSocketError {
    FailedToStartSocketError,
@@ -37,7 +155,6 @@ impl From<io::Error> for WebSocketError {
            io::ErrorKind::ConnectionReset => FailedToStartSocketError,
            io::ErrorKind::ConnectionAborted => FailedToStartSocketError,
            io::ErrorKind::NotConnected => FailedToStartSocketError,
-
            io::ErrorKind::AddrInUse => FailedToStartSocketError,
            io::ErrorKind::AddrNotAvailable => FailedToStartSocketError,
            _ => UnknownSocketError,
@@ -57,15 +174,11 @@ enum ClientRequest {
    OwnDetails,
 }

-impl From<Message> for ClientRequest {
-    fn from(msg: Message) -> Self {
-        let text_msg = match msg {
-            Message::Text(msg) => msg,
-            Message::Binary(_) => panic!("binary messages are not supported!"),
-            Message::Close(_) => panic!("todo: handle close!"),
-            _ => panic!("Other types of messages are also unsupported!"),
-        };
-        serde_json::from_str(&text_msg).expect("unable to deserialize From<Message> json")
+impl TryFrom<String> for ClientRequest {
+    type Error = serde_json::Error;
+
+    fn try_from(msg: String) -> Result<Self, Self::Error> {
+        serde_json::from_str(&msg)
    }
 }

@@ -75,6 +188,26 @@ impl ClientRequest {
        recipient_address: String,
        mut input_tx: mpsc::UnboundedSender<InputMessage>,
    ) -> ServerResponse {
+        let message_bytes = msg.into_bytes();
+        // TODO: wait until 0.4.0 release to replace those constants with newly exposed
+        // sphinx::constants::MAXIMUM_PLAINTEXT_LENGTH
+        // we can't do it now for compatibility reasons as most recent sphinx revision
+        // has breaking changes due to packet format changes
+        let maximum_plaintext_length = sphinx::constants::PAYLOAD_SIZE
+            - sphinx::constants::SECURITY_PARAMETER
+            - sphinx::constants::DESTINATION_ADDRESS_LENGTH
+            - 1;
+        if message_bytes.len() > maximum_plaintext_length {
+            return ServerResponse::Error {
+                message: format!(
+                    "too long message. Sent {} bytes while the maximum is {}",
+                    message_bytes.len(),
+                    maximum_plaintext_length
+                )
+                .to_string(),
+            };
+        }
+
        let address_vec = match base64::decode_config(&recipient_address, base64::URL_SAFE) {
            Err(e) => {
                return ServerResponse::Error {
@@ -95,9 +228,7 @@ impl ClientRequest {

        let dummy_surb = [0; 16];

-        let input_msg = InputMessage(Destination::new(address, dummy_surb), msg.into_bytes());
-
-        println!("ALMOST ABOUT TO SOMEDAY SEND {:?}", input_msg);
+        let input_msg = InputMessage(Destination::new(address, dummy_surb), message_bytes);
        input_tx.send(input_msg).await.unwrap();

        ServerResponse::Send
@@ -113,7 +244,6 @@ impl ClientRequest {
        }

        let messages = res_rx.map(|msg| msg).await;
-
        if messages.is_err() {
            warn!("Failed to handle_fetch. messages is an error");
            return ServerResponse::Error {
@@ -121,14 +251,22 @@ impl ClientRequest {
            };
        }

-        let messages = messages.unwrap();
-        let messages_as_b64 = messages
-            .iter()
-            .map(|message| base64::encode_config(message, base64::URL_SAFE))
+        let messages = messages
+            .unwrap()
+            .into_iter()
+            .map(|message| {
+                match std::str::from_utf8(&message) {
+                    Ok(v) => v,
+                    Err(e) => {
+                        error!("Invalid UTF-8 sequence in response message: {}", e);
+                        ""
+                    }
+                }
+                .to_owned()
+            })
            .collect();
-        ServerResponse::Fetch {
-            messages: messages_as_b64,
-        }
+
+        ServerResponse::Fetch { messages }
    }

    async fn handle_get_clients(topology: Topology) -> ServerResponse {
@@ -161,37 +299,13 @@ enum ServerResponse {

 impl Into<Message> for ServerResponse {
    fn into(self) -> Message {
+        // it should be safe to call `unwrap` here as the message is generated by the server
+        // so if it fails (and consequently panics) it's a bug that should be resolved
        let str_res = serde_json::to_string(&self).unwrap();
        Message::Text(str_res)
    }
 }

-async fn handle_connection(conn: Connection) {
-    let mut conn = conn;
-    while let Some(msg) = conn.rx.next().await {
-        println!("Received a message from {}: {}", conn.address, msg);
-        let request: ClientRequest = msg.into();
-
-        let response = match request {
-            ClientRequest::Send {
-                message,
-                recipient_address,
-            } => {
-                ClientRequest::handle_send(message, recipient_address, conn.msg_input.clone()).await
-            }
-            ClientRequest::Fetch => ClientRequest::handle_fetch(conn.msg_query.clone()).await,
-            ClientRequest::GetClients => {
-                ClientRequest::handle_get_clients(conn.topology.clone()).await
-            }
-            ClientRequest::OwnDetails => ClientRequest::handle_own_details(conn.self_address).await,
-        };
-
-        conn.tx
-            .unbounded_send(response.into())
-            .expect("Failed to forward message");
-    }
-}
-
 async fn accept_connection(
    stream: tokio::net::TcpStream,
    msg_input: mpsc::UnboundedSender<InputMessage>,
@@ -203,14 +317,12 @@ async fn accept_connection(
    let address = stream
        .peer_addr()
        .expect("connected streams should have a peer address");
-    println!("Peer address: {}", address);
+    debug!("Peer address: {}", address);

    let mut ws_stream = tokio_tungstenite::accept_async(stream)
        .await
        .expect("Error during the websocket handshake occurred");

-    println!("New WebSocket connection: {}", address);
-
    // Create a channel for our stream, which other sockets will use to
    // send us messages. Then register our address with the stream to send
    // data to us.
@@ -225,15 +337,50 @@ async fn accept_connection(
        msg_query,
        self_address,
    };
-    tokio::spawn(handle_connection(conn));
+    tokio::spawn(conn.handle());

    while let Some(message) = ws_stream.next().await {
-        let message = message.expect("Failed to get request");
-        msg_tx
-            .unbounded_send(message)
-            .expect("Failed to forward request");
+        let message = match message {
+            Ok(msg) => msg,
+            Err(err) => {
+                error!("failed to obtain message from websocket stream! stopping connection handler: {}", err);
+                return;
+            }
+        };
+
+        let mut should_close = false;
+        if message.is_close() {
+            should_close = true;
+        }
+
+        if let Err(err) = msg_tx.unbounded_send(message) {
+            error!(
+                "Failed to forward request. Closing the socket connection: {}",
+                err
+            );
+            return;
+        }
+        // if the received message is a close message it means we will reply with a close
+        // or it is a reply to our close, either way as per RFC6455 5.5.1 we should close the
+        // underlying TCP connection:
+        // After both sending and receiving a Close message, an endpoint
+        // considers the WebSocket connection closed and MUST close the
+        // underlying TCP connection. The server MUST close the underlying TCP
+        // connection immediately;
+
        if let Some(resp) = response_rx.next().await {
-            ws_stream.send(resp).await.expect("Failed to send response");
+            if let Err(err) = ws_stream.send(resp).await {
+                warn!(
+                    "Failed to send message over websocket: {}. Assuming the connection is dead.",
+                    err
+                );
+                return;
+            }
+        }
+
+        if should_close {
+            info!("Closing the websocket connection");
+            return;
        }
    }
 }
@@ -1,5 +1,9 @@
 # nym-sfw-provider Changelog

+## 0.3.3
+
+* Version increase for consistency with `nym-client`
+
 ## 0.3.2

 * Version increase for consistency with `nym-mixnode` and `nym-client`
@@ -1,7 +1,7 @@
 [package]
 build = "build.rs"
 name = "nym-sfw-provider"
-version = "0.3.2"
+version = "0.3.3"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
 edition = "2018"