diff --git a/.travis.yml b/.travis.yml index 1e59702b5f..7041ae8d65 100644 --- a/.travis.yml +++ b/.travis.yml @@ -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 \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index 9cc19b1dbe..6aa1d181a4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/README.md b/README.md index 347c14c636..d43dd40a96 100644 --- a/README.md +++ b/README.md @@ -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) diff --git a/mixnode/CHANGELOG.md b/mixnode/CHANGELOG.md index 2f41201d97..09fcd6f114 100644 --- a/mixnode/CHANGELOG.md +++ b/mixnode/CHANGELOG.md @@ -1,5 +1,9 @@ # nym-mixnode Changelog +## 0.3.3 + +* Version increase for consistency with `nym-client` + ## 0.3.2 * added separate announce address diff --git a/mixnode/Cargo.toml b/mixnode/Cargo.toml index 8c51ad61e0..45d800f1d8 100644 --- a/mixnode/Cargo.toml +++ b/mixnode/Cargo.toml @@ -1,7 +1,7 @@ [package] build = "build.rs" name = "nym-mixnode" -version = "0.3.2" +version = "0.3.3" authors = ["Dave Hrycyszyn ", "Jędrzej Stuczyński "] edition = "2018" diff --git a/nym-client/CHANGELOG.md b/nym-client/CHANGELOG.md index 34ed4c2dd2..10796961c0 100644 --- a/nym-client/CHANGELOG.md +++ b/nym-client/CHANGELOG.md @@ -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 diff --git a/nym-client/Cargo.toml b/nym-client/Cargo.toml index fbf8e54a62..f336d11b5a 100644 --- a/nym-client/Cargo.toml +++ b/nym-client/Cargo.toml @@ -1,12 +1,16 @@ [package] build = "build.rs" name = "nym-client" -version = "0.3.2" +version = "0.3.3" authors = ["Dave Hrycyszyn ", "Jędrzej Stuczyński "] 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" diff --git a/nym-client/src/built_info.rs b/nym-client/src/built_info.rs new file mode 100644 index 0000000000..d11fb6389f --- /dev/null +++ b/nym-client/src/built_info.rs @@ -0,0 +1,2 @@ +// The file has been placed there by the build script. +include!(concat!(env!("OUT_DIR"), "/built.rs")); diff --git a/nym-client/src/commands/init.rs b/nym-client/src/commands/init.rs index ce6ee879b4..f6d675ea02 100644 --- a/nym-client/src/commands/init.rs +++ b/nym-client/src/commands/init.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 diff --git a/nym-client/src/commands/tcpsocket.rs b/nym-client/src/commands/tcpsocket.rs index 97b258101e..3ed3662c7b 100644 --- a/nym-client/src/commands/tcpsocket.rs +++ b/nym-client/src/commands/tcpsocket.rs @@ -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::() { Ok(n) => n, diff --git a/nym-client/src/commands/websocket.rs b/nym-client/src/commands/websocket.rs index 4b27bc3455..99dbc1ecc6 100644 --- a/nym-client/src/commands/websocket.rs +++ b/nym-client/src/commands/websocket.rs @@ -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::() { Ok(n) => n, diff --git a/nym-client/src/lib.rs b/nym-client/src/lib.rs new file mode 100644 index 0000000000..5fa9078897 --- /dev/null +++ b/nym-client/src/lib.rs @@ -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; diff --git a/nym-client/src/main.rs b/nym-client/src/main.rs index b69fa95213..8455bc7287 100644 --- a/nym-client/src/main.rs +++ b/nym-client/src/main.rs @@ -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()); + } } } diff --git a/nym-client/src/sockets/ws.rs b/nym-client/src/sockets/ws.rs index 59bebfadbd..ab01fca011 100644 --- a/nym-client/src/sockets/ws.rs +++ b/nym-client/src/sockets/ws.rs @@ -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, } +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) -> 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) -> 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) -> 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>) -> 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 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 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 json") +impl TryFrom for ClientRequest { + type Error = serde_json::Error; + + fn try_from(msg: String) -> Result { + serde_json::from_str(&msg) } } @@ -75,6 +188,26 @@ impl ClientRequest { recipient_address: String, mut input_tx: mpsc::UnboundedSender, ) -> 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 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, @@ -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; } } } diff --git a/sfw-provider/CHANGELOG.md b/sfw-provider/CHANGELOG.md index 6da245a539..4c3c0773be 100644 --- a/sfw-provider/CHANGELOG.md +++ b/sfw-provider/CHANGELOG.md @@ -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` diff --git a/sfw-provider/Cargo.toml b/sfw-provider/Cargo.toml index 83bf161c9a..d76965a12f 100644 --- a/sfw-provider/Cargo.toml +++ b/sfw-provider/Cargo.toml @@ -1,7 +1,7 @@ [package] build = "build.rs" name = "nym-sfw-provider" -version = "0.3.2" +version = "0.3.3" authors = ["Dave Hrycyszyn ", "Jędrzej Stuczyński "] edition = "2018"