diff --git a/.gitignore b/.gitignore index dd3546d329..8262ad0500 100644 --- a/.gitignore +++ b/.gitignore @@ -7,4 +7,5 @@ target .env /.vscode/settings.json validator/.vscode -sample-configs/validator-config.toml \ No newline at end of file +sample-configs/validator-config.toml +.vscode \ No newline at end of file diff --git a/clients/desktop/examples/chunking.rs b/clients/desktop/examples/chunking.rs deleted file mode 100644 index 0e0d87f839..0000000000 --- a/clients/desktop/examples/chunking.rs +++ /dev/null @@ -1,69 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use config::NymConfig; -use std::time; -use tokio::runtime::Runtime; - -fn setup_logging() { - let mut log_builder = pretty_env_logger::formatted_timed_builder(); - if let Ok(s) = ::std::env::var("RUST_LOG") { - log_builder.parse_filters(&s); - } else { - // default to 'Info' - log_builder.filter(None, log::LevelFilter::Info); - } - - log_builder - .filter_module("hyper", log::LevelFilter::Warn) - .filter_module("tokio_reactor", log::LevelFilter::Warn) - .filter_module("reqwest", log::LevelFilter::Warn) - .filter_module("mio", log::LevelFilter::Warn) - .filter_module("want", log::LevelFilter::Warn) - .init(); -} - -fn main() { - // totally optional - setup_logging(); - - let mut rt = Runtime::new().unwrap(); - - // note: I have manually initialised the client with this config (because the keys weren't - // saved, etc - a really dodgy setup but good enough for a quick test) - // so to run it, you need to do the usual `nym-client init --id native-local` - // then optionally modify config.toml to increase rates, etc. if you want - let config = nym_client::config::Config::load_from_file(None, Some("native-local")).unwrap(); - let mut client = nym_client::client::NymClient::new(config); - - let input_data = std::fs::read("trailer.mp4").unwrap(); - - client.start(); - let address = client.as_mix_destination(); - client.send_message(address, input_data); - - loop { - let mut messages = rt.block_on(async { - tokio::time::delay_for(time::Duration::from_secs(2)).await; - client.check_for_messages_async().await - }); - - if !messages.is_empty() { - assert_eq!(messages.len(), 1); - std::fs::write("downloaded.mp4", messages.pop().unwrap()).unwrap(); - println!("done!"); - std::process::exit(0); - } - } -} diff --git a/clients/desktop/examples/dummy_file b/clients/desktop/examples/dummy_file new file mode 100644 index 0000000000..ee7e8a3e00 --- /dev/null +++ b/clients/desktop/examples/dummy_file @@ -0,0 +1,9 @@ +Lorem ipsum dolor sit amet, consectetur adipiscing elit. Nullam quis efficitur neque. Quisque aliquet vulputate ante, eget vehicula odio feugiat ac. Nulla ut mattis magna. Aenean tincidunt quis nulla eget eleifend. Cras in pretium sem. Nunc lorem metus, blandit sit amet egestas ut, feugiat quis tellus. Aenean tristique, enim a tincidunt condimentum, eros est blandit nunc, id viverra metus erat at nulla. Vivamus at tellus sodales, feugiat odio vel, laoreet neque. Vivamus posuere nulla ac sodales bibendum. + +Vestibulum pulvinar nisi non ultricies egestas. Integer finibus ultrices justo vitae suscipit. Etiam interdum eu justo vel interdum. Morbi sagittis ac nisl quis consequat. Mauris dapibus ut risus ac facilisis. Pellentesque non tortor feugiat, consectetur arcu vel, ullamcorper sapien. Proin sodales purus non orci bibendum, sit amet ultrices justo ullamcorper. Nullam ac risus ac justo ultricies efficitur auctor nec arcu. Etiam sed finibus felis. Suspendisse potenti. Phasellus malesuada velit ac ullamcorper egestas. Sed elementum diam ut est gravida ultricies. + +Pellentesque sed metus massa. Cras imperdiet lacus sit amet dolor aliquam, luctus posuere justo hendrerit. Morbi augue ex, gravida a metus sed, scelerisque euismod lacus. Nam consequat sapien ac pellentesque sagittis. Morbi a ultrices massa, vel aliquet ex. Maecenas ac sem diam. Nunc sed erat et ipsum volutpat auctor. Etiam elit felis, commodo vitae ipsum ac, fermentum lobortis arcu. Aliquam eu tempus enim. Curabitur vulputate imperdiet aliquam. Morbi iaculis rhoncus risus at malesuada. Donec accumsan feugiat ligula ut facilisis. Nunc porttitor sit amet est eget malesuada. Sed sed consectetur augue, non dapibus orci. Mauris aliquam pellentesque quam, sit amet pellentesque velit cursus vitae. Morbi sit amet molestie risus. + +Nam gravida non ligula a egestas. Fusce sodales, purus id rhoncus mattis, purus est vehicula urna, vel finibus augue velit et est. Donec dictum erat eleifend lobortis iaculis. Praesent id venenatis ante. Donec feugiat, ipsum eget porttitor pulvinar, nisl odio posuere lorem, ut placerat elit nulla a ligula. Suspendisse nec nibh tincidunt, sollicitudin mi a, volutpat ligula. In maximus quam lacus, eget semper dolor sagittis sit amet. + +In vitae hendrerit est, quis facilisis dui. In eu ante enim. Nullam hendrerit odio sit amet odio tincidunt eleifend. Aliquam erat volutpat. Curabitur commodo, purus pharetra lobortis rhoncus, tortor massa imperdiet nisl, vel dignissim tortor sem at orci. Aliquam maximus lobortis lacus, eu porttitor purus dapibus ut. Praesent at dapibus felis, efficitur blandit tortor. In hac habitasse platea dictumst. Aenean ultrices, nisl a pretium sagittis, tellus sapien mollis erat, eu consectetur erat mauris sed libero. Duis feugiat dapibus mi, vel ornare velit vehicula mattis. Ut suscipit pharetra leo et sollicitudin. \ No newline at end of file diff --git a/clients/desktop/examples/go-examples/websocket/dummy_file b/clients/desktop/examples/go-examples/websocket/dummy_file new file mode 100644 index 0000000000..ee7e8a3e00 --- /dev/null +++ b/clients/desktop/examples/go-examples/websocket/dummy_file @@ -0,0 +1,9 @@ +Lorem ipsum dolor sit amet, consectetur adipiscing elit. Nullam quis efficitur neque. Quisque aliquet vulputate ante, eget vehicula odio feugiat ac. Nulla ut mattis magna. Aenean tincidunt quis nulla eget eleifend. Cras in pretium sem. Nunc lorem metus, blandit sit amet egestas ut, feugiat quis tellus. Aenean tristique, enim a tincidunt condimentum, eros est blandit nunc, id viverra metus erat at nulla. Vivamus at tellus sodales, feugiat odio vel, laoreet neque. Vivamus posuere nulla ac sodales bibendum. + +Vestibulum pulvinar nisi non ultricies egestas. Integer finibus ultrices justo vitae suscipit. Etiam interdum eu justo vel interdum. Morbi sagittis ac nisl quis consequat. Mauris dapibus ut risus ac facilisis. Pellentesque non tortor feugiat, consectetur arcu vel, ullamcorper sapien. Proin sodales purus non orci bibendum, sit amet ultrices justo ullamcorper. Nullam ac risus ac justo ultricies efficitur auctor nec arcu. Etiam sed finibus felis. Suspendisse potenti. Phasellus malesuada velit ac ullamcorper egestas. Sed elementum diam ut est gravida ultricies. + +Pellentesque sed metus massa. Cras imperdiet lacus sit amet dolor aliquam, luctus posuere justo hendrerit. Morbi augue ex, gravida a metus sed, scelerisque euismod lacus. Nam consequat sapien ac pellentesque sagittis. Morbi a ultrices massa, vel aliquet ex. Maecenas ac sem diam. Nunc sed erat et ipsum volutpat auctor. Etiam elit felis, commodo vitae ipsum ac, fermentum lobortis arcu. Aliquam eu tempus enim. Curabitur vulputate imperdiet aliquam. Morbi iaculis rhoncus risus at malesuada. Donec accumsan feugiat ligula ut facilisis. Nunc porttitor sit amet est eget malesuada. Sed sed consectetur augue, non dapibus orci. Mauris aliquam pellentesque quam, sit amet pellentesque velit cursus vitae. Morbi sit amet molestie risus. + +Nam gravida non ligula a egestas. Fusce sodales, purus id rhoncus mattis, purus est vehicula urna, vel finibus augue velit et est. Donec dictum erat eleifend lobortis iaculis. Praesent id venenatis ante. Donec feugiat, ipsum eget porttitor pulvinar, nisl odio posuere lorem, ut placerat elit nulla a ligula. Suspendisse nec nibh tincidunt, sollicitudin mi a, volutpat ligula. In maximus quam lacus, eget semper dolor sagittis sit amet. + +In vitae hendrerit est, quis facilisis dui. In eu ante enim. Nullam hendrerit odio sit amet odio tincidunt eleifend. Aliquam erat volutpat. Curabitur commodo, purus pharetra lobortis rhoncus, tortor massa imperdiet nisl, vel dignissim tortor sem at orci. Aliquam maximus lobortis lacus, eu porttitor purus dapibus ut. Praesent at dapibus felis, efficitur blandit tortor. In hac habitasse platea dictumst. Aenean ultrices, nisl a pretium sagittis, tellus sapien mollis erat, eu consectetur erat mauris sed libero. Duis feugiat dapibus mi, vel ornare velit vehicula mattis. Ut suscipit pharetra leo et sollicitudin. \ No newline at end of file diff --git a/clients/desktop/examples/go-examples/websocket/filesend.go b/clients/desktop/examples/go-examples/websocket/filesend.go new file mode 100644 index 0000000000..0ab17c090e --- /dev/null +++ b/clients/desktop/examples/go-examples/websocket/filesend.go @@ -0,0 +1,70 @@ +package main + +import ( + "encoding/json" + "fmt" + "github.com/btcsuite/btcutil/base58" + "github.com/gorilla/websocket" + "io/ioutil" +) + +func getSelfAddress(conn *websocket.Conn) string { + selfAddressRequest, err := json.Marshal(map[string]string{"type": "selfAddress"}) + if err != nil { + panic(err) + } + + if err = conn.WriteMessage(websocket.TextMessage, []byte(selfAddressRequest)); err != nil { + panic(err) + } + + responseJSON := make(map[string]interface{}) + err = conn.ReadJSON(&responseJSON) + if err != nil { + panic(err) + } + + return responseJSON["address"].(string) +} + +func main() { + uri := "ws://localhost:1977" + + conn, _, err := websocket.DefaultDialer.Dial(uri, nil) + if err != nil { + panic(err) + } + defer conn.Close() + + selfAddress := getSelfAddress(conn) + fmt.Printf("our address is: %v\n", selfAddress) + decodedAddress := base58.Decode(selfAddress) + + read_data, err := ioutil.ReadFile("dummy_file") + if err != nil { + panic(err) + } + + payload := append(decodedAddress[:], read_data[:]...) + fmt.Printf("sending content of 'dummy file' over the mix network...\n") + if err = conn.WriteMessage(websocket.BinaryMessage, payload); err != nil { + panic(err) + } + sendConfirmationJSON := make(map[string]interface{}) + err = conn.ReadJSON(&sendConfirmationJSON) + if err != nil { + panic(err) + } + if sendConfirmationJSON["type"].(string) != "send" { + panic("invalid send confirmation") + } + + fmt.Printf("waiting to receive a message from the mix network...\n") + _, receivedMessage, err := conn.ReadMessage() + if err != nil { + panic(err) + } + + fmt.Printf("writing the file back to the disk!\n") + ioutil.WriteFile("received_file", receivedMessage, 0644) +} diff --git a/clients/desktop/examples/go-examples/websocket/textsend.go b/clients/desktop/examples/go-examples/websocket/textsend.go new file mode 100644 index 0000000000..2f46490016 --- /dev/null +++ b/clients/desktop/examples/go-examples/websocket/textsend.go @@ -0,0 +1,67 @@ +package main + +import ( + "encoding/json" + "fmt" + + "github.com/gorilla/websocket" +) + +func getSelfAddress(conn *websocket.Conn) string { + selfAddressRequest, err := json.Marshal(map[string]string{"type": "selfAddress"}) + if err != nil { + panic(err) + } + + if err = conn.WriteMessage(websocket.TextMessage, []byte(selfAddressRequest)); err != nil { + panic(err) + } + + responseJSON := make(map[string]interface{}) + err = conn.ReadJSON(&responseJSON) + if err != nil { + panic(err) + } + + return responseJSON["address"].(string) +} + +func main() { + message := "Hello Nym!" + + uri := "ws://localhost:1977" + + conn, _, err := websocket.DefaultDialer.Dial(uri, nil) + if err != nil { + panic(err) + } + defer conn.Close() + + selfAddress := getSelfAddress(conn) + fmt.Printf("our address is: %v\n", selfAddress) + sendRequest, err := json.Marshal(map[string]string{"type": "send", "recipient": selfAddress, "message": message}) + if err != nil { + panic(err) + } + + fmt.Printf("sending '%v' over the mix network...\n", message) + if err = conn.WriteMessage(websocket.TextMessage, []byte(sendRequest)); err != nil { + panic(err) + } + + sendConfirmationJSON := make(map[string]interface{}) + err = conn.ReadJSON(&sendConfirmationJSON) + if err != nil { + panic(err) + } + if sendConfirmationJSON["type"].(string) != "send" { + panic("invalid send confirmation") + } + + fmt.Printf("waiting to receive a message from the mix network...\n") + _, receivedMessage, err := conn.ReadMessage() + if err != nil { + panic(err) + } + fmt.Printf("received %v from the mix network!\n", string(receivedMessage)) +} diff --git a/clients/desktop/js-examples/websocket/.gitignore b/clients/desktop/examples/js-examples/websocket/.gitignore similarity index 100% rename from clients/desktop/js-examples/websocket/.gitignore rename to clients/desktop/examples/js-examples/websocket/.gitignore diff --git a/clients/desktop/js-examples/websocket/README.md b/clients/desktop/examples/js-examples/websocket/README.md similarity index 100% rename from clients/desktop/js-examples/websocket/README.md rename to clients/desktop/examples/js-examples/websocket/README.md diff --git a/clients/desktop/js-examples/websocket/package-lock.json b/clients/desktop/examples/js-examples/websocket/package-lock.json similarity index 100% rename from clients/desktop/js-examples/websocket/package-lock.json rename to clients/desktop/examples/js-examples/websocket/package-lock.json diff --git a/clients/desktop/js-examples/websocket/package.json b/clients/desktop/examples/js-examples/websocket/package.json similarity index 100% rename from clients/desktop/js-examples/websocket/package.json rename to clients/desktop/examples/js-examples/websocket/package.json diff --git a/clients/desktop/js-examples/websocket/src/index.html b/clients/desktop/examples/js-examples/websocket/src/index.html similarity index 100% rename from clients/desktop/js-examples/websocket/src/index.html rename to clients/desktop/examples/js-examples/websocket/src/index.html diff --git a/clients/desktop/js-examples/websocket/src/index.js b/clients/desktop/examples/js-examples/websocket/src/index.js similarity index 100% rename from clients/desktop/js-examples/websocket/src/index.js rename to clients/desktop/examples/js-examples/websocket/src/index.js diff --git a/clients/desktop/js-examples/websocket/webpack.config.js b/clients/desktop/examples/js-examples/websocket/webpack.config.js similarity index 100% rename from clients/desktop/js-examples/websocket/webpack.config.js rename to clients/desktop/examples/js-examples/websocket/webpack.config.js diff --git a/clients/desktop/examples/python-examples/websocket/dummy_file b/clients/desktop/examples/python-examples/websocket/dummy_file new file mode 100644 index 0000000000..ee7e8a3e00 --- /dev/null +++ b/clients/desktop/examples/python-examples/websocket/dummy_file @@ -0,0 +1,9 @@ +Lorem ipsum dolor sit amet, consectetur adipiscing elit. Nullam quis efficitur neque. Quisque aliquet vulputate ante, eget vehicula odio feugiat ac. Nulla ut mattis magna. Aenean tincidunt quis nulla eget eleifend. Cras in pretium sem. Nunc lorem metus, blandit sit amet egestas ut, feugiat quis tellus. Aenean tristique, enim a tincidunt condimentum, eros est blandit nunc, id viverra metus erat at nulla. Vivamus at tellus sodales, feugiat odio vel, laoreet neque. Vivamus posuere nulla ac sodales bibendum. + +Vestibulum pulvinar nisi non ultricies egestas. Integer finibus ultrices justo vitae suscipit. Etiam interdum eu justo vel interdum. Morbi sagittis ac nisl quis consequat. Mauris dapibus ut risus ac facilisis. Pellentesque non tortor feugiat, consectetur arcu vel, ullamcorper sapien. Proin sodales purus non orci bibendum, sit amet ultrices justo ullamcorper. Nullam ac risus ac justo ultricies efficitur auctor nec arcu. Etiam sed finibus felis. Suspendisse potenti. Phasellus malesuada velit ac ullamcorper egestas. Sed elementum diam ut est gravida ultricies. + +Pellentesque sed metus massa. Cras imperdiet lacus sit amet dolor aliquam, luctus posuere justo hendrerit. Morbi augue ex, gravida a metus sed, scelerisque euismod lacus. Nam consequat sapien ac pellentesque sagittis. Morbi a ultrices massa, vel aliquet ex. Maecenas ac sem diam. Nunc sed erat et ipsum volutpat auctor. Etiam elit felis, commodo vitae ipsum ac, fermentum lobortis arcu. Aliquam eu tempus enim. Curabitur vulputate imperdiet aliquam. Morbi iaculis rhoncus risus at malesuada. Donec accumsan feugiat ligula ut facilisis. Nunc porttitor sit amet est eget malesuada. Sed sed consectetur augue, non dapibus orci. Mauris aliquam pellentesque quam, sit amet pellentesque velit cursus vitae. Morbi sit amet molestie risus. + +Nam gravida non ligula a egestas. Fusce sodales, purus id rhoncus mattis, purus est vehicula urna, vel finibus augue velit et est. Donec dictum erat eleifend lobortis iaculis. Praesent id venenatis ante. Donec feugiat, ipsum eget porttitor pulvinar, nisl odio posuere lorem, ut placerat elit nulla a ligula. Suspendisse nec nibh tincidunt, sollicitudin mi a, volutpat ligula. In maximus quam lacus, eget semper dolor sagittis sit amet. + +In vitae hendrerit est, quis facilisis dui. In eu ante enim. Nullam hendrerit odio sit amet odio tincidunt eleifend. Aliquam erat volutpat. Curabitur commodo, purus pharetra lobortis rhoncus, tortor massa imperdiet nisl, vel dignissim tortor sem at orci. Aliquam maximus lobortis lacus, eu porttitor purus dapibus ut. Praesent at dapibus felis, efficitur blandit tortor. In hac habitasse platea dictumst. Aenean ultrices, nisl a pretium sagittis, tellus sapien mollis erat, eu consectetur erat mauris sed libero. Duis feugiat dapibus mi, vel ornare velit vehicula mattis. Ut suscipit pharetra leo et sollicitudin. \ No newline at end of file diff --git a/clients/desktop/examples/python-examples/websocket/filesend.py b/clients/desktop/examples/python-examples/websocket/filesend.py new file mode 100644 index 0000000000..4df3e12ff0 --- /dev/null +++ b/clients/desktop/examples/python-examples/websocket/filesend.py @@ -0,0 +1,35 @@ + +import asyncio +import base58 +import json +import websockets + +self_address_request = json.dumps({ + "type": "selfAddress" +}) + + +async def send_file(): + uri = "ws://localhost:1977" + async with websockets.connect(uri) as websocket: + await websocket.send(self_address_request) + self_address = json.loads(await websocket.recv()) + print("our address is: {}".format(self_address["address"])) + + bin_payload = bytearray(base58.b58decode(self_address["address"])) + with open("dummy_file", "rb") as input_file: + read_data = input_file.read() + bin_payload += read_data + + print("sending content of 'dummy_file' over the mix network...") + await websocket.send(bin_payload) + msg_send_confirmation = json.loads(await websocket.recv()) + assert msg_send_confirmation["type"], "send" + + print("waiting to receive the 'dummy_file' from the mix network...") + received_data = await websocket.recv() + with open("received_file", "wb") as output_file: + print("writing the file back to the disk!") + output_file.write(received_data) + +asyncio.get_event_loop().run_until_complete(send_file()) diff --git a/clients/desktop/examples/python-examples/websocket/textsend.py b/clients/desktop/examples/python-examples/websocket/textsend.py new file mode 100644 index 0000000000..972d8bc1c6 --- /dev/null +++ b/clients/desktop/examples/python-examples/websocket/textsend.py @@ -0,0 +1,36 @@ + +import asyncio +import base58 +import json +import websockets + +self_address_request = json.dumps({ + "type": "selfAddress" +}) + + +async def send_file(): + message = "Hello Nym!" + + uri = "ws://localhost:1977" + async with websockets.connect(uri) as websocket: + await websocket.send(self_address_request) + self_address = json.loads(await websocket.recv()) + print("our address is: {}".format(self_address["address"])) + + text_send = json.dumps({ + "type": "send", + "message": message, + "recipient": self_address["address"] + }) + + print("sending '{}' over the mix network...".format(message)) + await websocket.send(text_send) + msg_send_confirmation = json.loads(await websocket.recv()) + assert msg_send_confirmation["type"], "send" + + print("waiting to receive a message from the mix network...") + received_message = await websocket.recv() + print("received {} from the mix network!".format(received_message)) + +asyncio.get_event_loop().run_until_complete(send_file()) diff --git a/clients/desktop/examples/websocket_filesend.rs b/clients/desktop/examples/websocket_filesend.rs new file mode 100644 index 0000000000..239bddf7ec --- /dev/null +++ b/clients/desktop/examples/websocket_filesend.rs @@ -0,0 +1,62 @@ +use bs58; +use futures::{SinkExt, StreamExt}; +use nym_client::websocket::{BinaryClientRequest, ClientRequest, ServerResponse}; +use nymsphinx::DestinationAddressBytes; +use std::convert::TryFrom; +use tokio_tungstenite::{connect_async, tungstenite::protocol::Message}; +#[tokio::main] +async fn main() { + let uri = "ws://localhost:1977"; + let (mut ws_stream, _) = connect_async(uri).await.unwrap(); + + let self_address_request = ClientRequest::SelfAddress; + ws_stream.send(self_address_request.into()).await.unwrap(); + + let raw_response = ws_stream.next().await.unwrap().unwrap(); + // what we received now is just a json, but we know it's exact format + // so might as well use that + let response = match raw_response { + Message::Text(txt_msg) => ServerResponse::try_from(txt_msg).unwrap(), + _ => panic!("received an unexpected response type!"), + }; + + let self_address = match response { + ServerResponse::SelfAddress { address } => address, + _ => panic!("received an unexpected response type!"), + }; + println!("our address is: {}", self_address.clone()); + + let mut address_bytes = [0; 32]; + bs58::decode(self_address).into(&mut address_bytes).unwrap(); + + // this is equivalent to just sending bs58 decoding of the address, which is always 32 bytes + let decoded_address = DestinationAddressBytes::from_bytes(address_bytes); + let read_data = std::fs::read("examples/dummy_file").unwrap(); + + let send_request = BinaryClientRequest::Send { + recipient_address: decoded_address, + data: read_data, + }; + + println!("sending content of 'dummy_file' over the mix network..."); + ws_stream.send(send_request.into()).await.unwrap(); + + let raw_send_confirmation = ws_stream.next().await.unwrap().unwrap(); + let _send_confirmation = match raw_send_confirmation { + Message::Text(txt_msg) => match ServerResponse::try_from(txt_msg).unwrap() { + ServerResponse::Send => (), + _ => panic!("received an unexpected response type!"), + }, + _ => panic!("received an unexpected response type!"), + }; + + println!("waiting to receive the 'dummy_file' from the mix network..."); + let raw_message = ws_stream.next().await.unwrap().unwrap(); + let message = match raw_message { + Message::Binary(bin_payload) => bin_payload, + _ => panic!("received an unexpected response type!"), + }; + + println!("writing the file back to the disk!"); + std::fs::write("examples/received_file", message).unwrap(); +} diff --git a/clients/desktop/examples/websocket_textsend.rs b/clients/desktop/examples/websocket_textsend.rs new file mode 100644 index 0000000000..8d9b3a07eb --- /dev/null +++ b/clients/desktop/examples/websocket_textsend.rs @@ -0,0 +1,53 @@ +use futures::{SinkExt, StreamExt}; +use nym_client::websocket::{ClientRequest, ServerResponse}; +use std::convert::TryFrom; +use tokio_tungstenite::{connect_async, tungstenite::protocol::Message}; +#[tokio::main] +async fn main() { + let message = "Hello Nym!".to_string(); + + let uri = "ws://localhost:1977"; + let (mut ws_stream, _) = connect_async(uri).await.unwrap(); + + let self_address_request = ClientRequest::SelfAddress; + ws_stream.send(self_address_request.into()).await.unwrap(); + + let raw_response = ws_stream.next().await.unwrap().unwrap(); + // what we received now is just a json, but we know it's exact format + // so might as well use that + let response = match raw_response { + Message::Text(txt_msg) => ServerResponse::try_from(txt_msg).unwrap(), + _ => panic!("received an unexpected response type!"), + }; + + let self_address = match response { + ServerResponse::SelfAddress { address } => address, + _ => panic!("received an unexpected response type!"), + }; + println!("our address is: {}", self_address.clone()); + + let send_request = ClientRequest::Send { + message: message.clone(), + recipient: self_address, + }; + println!("sending {:?} over the mix network...", message); + ws_stream.send(send_request.into()).await.unwrap(); + + let raw_send_confirmation = ws_stream.next().await.unwrap().unwrap(); + let _send_confirmation = match raw_send_confirmation { + Message::Text(txt_msg) => match ServerResponse::try_from(txt_msg).unwrap() { + ServerResponse::Send => (), + _ => panic!("received an unexpected response type!"), + }, + _ => panic!("received an unexpected response type!"), + }; + + println!("waiting to receive a message from the mix network..."); + let raw_message = ws_stream.next().await.unwrap().unwrap(); + let message = match raw_message { + Message::Text(txt_msg) => txt_msg, + _ => panic!("received an unexpected response type!"), + }; + + println!("received {:?} from the mix network!", message); +} diff --git a/clients/desktop/src/client/mod.rs b/clients/desktop/src/client/mod.rs index 05b86772ed..bae33a7149 100644 --- a/clients/desktop/src/client/mod.rs +++ b/clients/desktop/src/client/mod.rs @@ -22,16 +22,17 @@ use crate::client::topology_control::{ }; use crate::config::persistence::pathfinder::ClientPathfinder; use crate::config::{Config, SocketType}; -use crate::sockets::{tcp, websocket}; +use crate::websocket; use crypto::identity::MixIdentityKeyPair; use directory_client::presence; -use futures::channel::{mpsc, oneshot}; +use futures::channel::mpsc; use gateway_client::{GatewayClient, SphinxPacketReceiver, SphinxPacketSender}; use gateway_requests::auth_token::AuthToken; use log::*; use nymsphinx::chunking::split_and_prepare_payloads; use nymsphinx::{Destination, DestinationAddressBytes}; use pemstore::pemstore::PemStore; +use received_buffer::{ReceivedBufferMessage, ReconstructeredMessagesReceiver}; use tokio::runtime::Runtime; use topology::NymTopology; @@ -53,11 +54,12 @@ pub struct NymClient { input_tx: Option, // to be used by "receive" function or socket, etc - receive_tx: Option, + receive_tx: Option, } #[derive(Debug)] // TODO: make fields private +// TODO2: make it take just destination address, because we don't care about SURBs (in this form) pub(crate) struct InputMessage(pub Destination, pub Vec); impl NymClient { @@ -261,35 +263,23 @@ impl NymClient { MixTrafficController::new(mix_rx, gateway_client).start(self.runtime.handle()); } - fn start_socket_listener( + fn start_websocket_listener( &self, topology_accessor: TopologyAccessor, - received_messages_buffer_output_tx: ReceivedBufferRequestSender, - input_tx: InputMessageSender, + buffer_requester: ReceivedBufferRequestSender, + msg_input: InputMessageSender, ) { - match self.config.get_socket_type() { - SocketType::WebSocket => { - websocket::listener::run( - self.runtime.handle(), - self.config.get_listening_port(), - input_tx, - received_messages_buffer_output_tx, - self.identity_keypair.public_key().derive_address(), - topology_accessor, - ); - } - SocketType::TCP => { - tcp::start_tcpsocket( - self.runtime.handle(), - self.config.get_listening_port(), - input_tx, - received_messages_buffer_output_tx, - self.identity_keypair.public_key().derive_address(), - topology_accessor, - ); - } - SocketType::None => (), - } + info!("Starting 'websocket listener'..."); + + let websocket_handler = websocket::Handler::new( + msg_input, + buffer_requester, + self.as_mix_destination_address(), + topology_accessor, + ); + + websocket::Listener::new(self.config.get_listening_port()) + .start(self.runtime.handle(), websocket_handler); } /// EXPERIMENTAL DIRECT RUST API @@ -314,15 +304,17 @@ impl NymClient { /// EXPERIMENTAL DIRECT RUST API /// It's untested and there are absolutely no guarantees about it (but seems to have worked /// well enough in local tests) - pub async fn check_for_messages_async(&self) -> Vec> { - let (res_tx, res_rx) = oneshot::channel(); - self.receive_tx - .as_ref() - .expect("start method was not called before!") - .unbounded_send(res_tx) - .unwrap(); + /// Note: it waits for the first occurence of messages being sent to ourselves. If you expect multiple + /// messages, you might have to call this function repeatedly. + pub async fn wait_for_messages(&mut self) -> Vec> { + use futures::StreamExt; - res_rx.await.unwrap() + self.receive_tx + .as_mut() + .expect("start method was not called before!") + .next() + .await + .expect("buffer controller seems to have somehow died!") } /// blocking version of `start` method. Will run forever (or until SIGINT is sent) @@ -343,24 +335,24 @@ impl NymClient { pub fn start(&mut self) { info!("Starting nym client"); // channels for inter-component communication + // TODO: make the channels be internally created by the relevant components + // rather than creating them here, so say for example the buffer controller would create the request channels + // and would allow anyone to clone the sender channel - // mix_tx is the transmitter for any component generating sphinx packets that are to be sent to the mixnet + // sphinx_message_sender is the transmitter for any component generating sphinx packets that are to be sent to the mixnet // they are used by cover traffic stream and real traffic stream - // mix_rx is the receiver used by MixTrafficController that sends the actual traffic - let (mix_tx, mix_rx) = mpsc::unbounded(); + // sphinx_message_receiver is the receiver used by MixTrafficController that sends the actual traffic + let (sphinx_message_sender, sphinx_message_receiver) = mpsc::unbounded(); - // sphinx_packet_tx is the transmitter of sphinx messages received from the gateway - // sphinx_packet_rx is the receiver for said messages - used by ReceivedMessagesBuffer - let (sphinx_packet_tx, sphinx_packet_rx) = mpsc::unbounded(); + // unwrapped_sphinx_sender is the transmitter of [unwrapped] sphinx messages received from the gateway + // unwrapped_sphinx_receiver is the receiver for said messages - used by ReceivedMessagesBuffer + let (unwrapped_sphinx_sender, unwrapped_sphinx_receiver) = mpsc::unbounded(); - // received_messages_buffer_output_tx is the transmitter for *REQUESTS* for messages contained in ReceivedMessagesBuffer - used by sockets - // the requests contain a oneshot channel to send a reply on - // received_messages_buffer_output_rx is the received for the said requests - used by ReceivedMessagesBuffer - let (received_messages_buffer_output_tx, received_messages_buffer_output_rx) = - mpsc::unbounded(); + // used for annoucing connectiong or disconnection of a channel for pushing re-assembled messages to + let (received_buffer_request_sender, received_buffer_request_receiver) = mpsc::unbounded(); // channels responsible for controlling real messages - let (input_tx, input_rx) = mpsc::unbounded::(); + let (input_sender, input_receiver) = mpsc::unbounded::(); // TODO: when we switch to our graph topology, we need to remember to change 'presence::Topology' type let shared_topology_accessor = TopologyAccessor::::new(); @@ -368,26 +360,49 @@ impl NymClient { // do not change that. self.start_topology_refresher(shared_topology_accessor.clone()); self.start_received_messages_buffer_controller( - received_messages_buffer_output_rx, - sphinx_packet_rx, + received_buffer_request_receiver, + unwrapped_sphinx_receiver, ); let gateway_url = self.runtime.block_on(Self::get_gateway_address( self.config.get_gateway_id(), shared_topology_accessor.clone(), )); - let gateway_client = self.start_gateway_client(sphinx_packet_tx, gateway_url); + let gateway_client = self.start_gateway_client(unwrapped_sphinx_sender, gateway_url); - self.start_mix_traffic_controller(mix_rx, gateway_client); - self.start_cover_traffic_stream(shared_topology_accessor.clone(), mix_tx.clone()); - self.start_real_traffic_stream(shared_topology_accessor.clone(), mix_tx, input_rx); - self.start_socket_listener( - shared_topology_accessor, - received_messages_buffer_output_tx.clone(), - input_tx.clone(), + self.start_mix_traffic_controller(sphinx_message_receiver, gateway_client); + self.start_cover_traffic_stream( + shared_topology_accessor.clone(), + sphinx_message_sender.clone(), ); - self.input_tx = Some(input_tx); - self.receive_tx = Some(received_messages_buffer_output_tx); + self.start_real_traffic_stream( + shared_topology_accessor.clone(), + sphinx_message_sender, + input_receiver, + ); + + match self.config.get_socket_type() { + SocketType::WebSocket => self.start_websocket_listener( + shared_topology_accessor, + received_buffer_request_sender, + input_sender.clone(), + ), + SocketType::None => { + // if we did not start the socket, it means we're running (supposedly) in the native mode + // and hence we should announce 'ourselves' to the buffer + let (reconstructed_sender, reconstructed_receiver) = mpsc::unbounded(); + + // tell the buffer to start sending stuff to us + received_buffer_request_sender + .unbounded_send(ReceivedBufferMessage::ReceiverAnnounce( + reconstructed_sender, + )) + .expect("the buffer request failed!"); + + self.receive_tx = Some(reconstructed_receiver); + self.input_tx = Some(input_sender); + } + } info!("Client startup finished!"); } diff --git a/clients/desktop/src/client/received_buffer.rs b/clients/desktop/src/client/received_buffer.rs index 4781ba4613..3e11edd530 100644 --- a/clients/desktop/src/client/received_buffer.rs +++ b/clients/desktop/src/client/received_buffer.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use futures::channel::{mpsc, oneshot}; +use futures::channel::mpsc; use futures::lock::Mutex; use futures::StreamExt; use gateway_client::SphinxPacketReceiver; @@ -23,13 +23,19 @@ use std::sync::Arc; use tokio::runtime::Handle; use tokio::task::JoinHandle; -pub(crate) type ReceivedBufferResponse = oneshot::Sender>>; -pub(crate) type ReceivedBufferRequestSender = mpsc::UnboundedSender; -pub(crate) type ReceivedBufferRequestReceiver = mpsc::UnboundedReceiver; +// Buffer Requests to say "hey, send any reconstructed messages to this channel" +// or to say "hey, I'm going offline, don't send anything more to me. Just buffer them instead" +pub(crate) type ReceivedBufferRequestSender = mpsc::UnboundedSender; +pub(crate) type ReceivedBufferRequestReceiver = mpsc::UnboundedReceiver; + +// The channel set for the above +pub(crate) type ReconstructeredMessagesSender = mpsc::UnboundedSender>>; +pub(crate) type ReconstructeredMessagesReceiver = mpsc::UnboundedReceiver>>; struct ReceivedMessagesBufferInner { messages: Vec>, message_reconstructor: MessageReconstructor, + message_sender: Option, } #[derive(Debug, Clone)] @@ -45,10 +51,49 @@ impl ReceivedMessagesBuffer { inner: Arc::new(Mutex::new(ReceivedMessagesBufferInner { messages: Vec::new(), message_reconstructor: MessageReconstructor::new(), + message_sender: None, })), } } + async fn disconnect_sender(&mut self) { + let mut guard = self.inner.lock().await; + if guard.message_sender.is_none() { + // in theory we could just ignore it, but that situation should have never happened + // in the first place, so this way we at least know we have an important bug to fix + panic!("trying to disconnect non-existent sender!") + } + guard.message_sender = None; + } + + async fn connect_sender(&mut self, sender: ReconstructeredMessagesSender) { + let mut guard = self.inner.lock().await; + if guard.message_sender.is_some() { + // in theory we could just ignore it, but that situation should have never happened + // in the first place, so this way we at least know we have an important bug to fix + panic!("trying overwrite an existing sender!") + } + + // while we're at it, also empty the buffer if we happened to receive anything while + // no sender was connected + let stored_messages = std::mem::replace(&mut guard.messages, Vec::new()); + if !stored_messages.is_empty() { + if let Err(err) = sender.unbounded_send(stored_messages) { + error!( + "The sender channel we just received is already invalidated - {:?}", + err + ); + // put the values back to the buffer + // the returned error has two fields: err: SendError and val: T, + // where val is the value that was failed to get sent; + // it's returned by the `into_inner` call + guard.messages = err.into_inner(); + return; + } + } + guard.message_sender = Some(sender); + } + async fn add_reconstructed_messages(&mut self, msgs: Vec>) { debug!("Adding {:?} new messages to the buffer!", msgs.len()); trace!("Adding new messages to the buffer! {:?}", msgs); @@ -76,18 +121,36 @@ impl ReceivedMessagesBuffer { completed_messages.push(reconstructed_message); } } - // make sure to drop the lock to not deadlock - drop(inner_guard); + if !completed_messages.is_empty() { - self.add_reconstructed_messages(completed_messages).await; + if let Some(sender) = &inner_guard.message_sender { + trace!("Sending reconstructed messages to announced sender"); + if let Err(err) = sender.unbounded_send(completed_messages) { + warn!("The reconstructed message receiver went offline without explicit notification (relevant error: - {:?})", err); + // make sure to drop the lock to not deadlock + // (it is required by `add_reconstructed_messages`) + inner_guard.message_sender = None; + drop(inner_guard); + self.add_reconstructed_messages(err.into_inner()).await; + } + } else { + // make sure to drop the lock to not deadlock + // (it is required by `add_reconstructed_messages`) + drop(inner_guard); + trace!("No sender available - buffering reconstructed messages"); + self.add_reconstructed_messages(completed_messages).await; + } } } +} - async fn acquire_and_empty(&mut self) -> Vec> { - trace!("Emptying the buffer and returning all messages"); - let mut mutex_guard = self.inner.lock().await; - std::mem::replace(&mut mutex_guard.messages, Vec::new()) - } +pub(crate) enum ReceivedBufferMessage { + // Signals a websocket connection (or a native implementation) was established and we should stop buffering messages, + // and instead send them directly to the received channel + ReceiverAnnounce(ReconstructeredMessagesSender), + + // Explicit signal that Receiver connection will no longer accept messages + ReceiverDisconnect, } struct RequestReceiver { @@ -109,29 +172,30 @@ impl RequestReceiver { fn start(mut self, handle: &Handle) -> JoinHandle<()> { handle.spawn(async move { while let Some(request) = self.query_receiver.next().await { - let messages = self.received_buffer.acquire_and_empty().await; - if let Err(failed_messages) = request.send(messages) { - error!( - "Failed to send the messages to the requester. Adding them back to the buffer" - ); - self.received_buffer.add_reconstructed_messages(failed_messages).await; + match request { + ReceivedBufferMessage::ReceiverAnnounce(sender) => { + self.received_buffer.connect_sender(sender).await; + } + ReceivedBufferMessage::ReceiverDisconnect => { + self.received_buffer.disconnect_sender().await + } } } }) } } -struct MessageReceiver { +struct FragmentedMessageReceiver { received_buffer: ReceivedMessagesBuffer, sphinx_packet_receiver: SphinxPacketReceiver, } -impl MessageReceiver { +impl FragmentedMessageReceiver { fn new( received_buffer: ReceivedMessagesBuffer, sphinx_packet_receiver: SphinxPacketReceiver, ) -> Self { - MessageReceiver { + FragmentedMessageReceiver { received_buffer, sphinx_packet_receiver, } @@ -148,7 +212,7 @@ impl MessageReceiver { } pub(crate) struct ReceivedMessagesBufferController { - messsage_receiver: MessageReceiver, + fragmented_messsage_receiver: FragmentedMessageReceiver, request_receiver: RequestReceiver, } @@ -160,7 +224,7 @@ impl ReceivedMessagesBufferController { let received_buffer = ReceivedMessagesBuffer::new(); ReceivedMessagesBufferController { - messsage_receiver: MessageReceiver::new( + fragmented_messsage_receiver: FragmentedMessageReceiver::new( received_buffer.clone(), sphinx_packet_receiver, ), @@ -170,7 +234,7 @@ impl ReceivedMessagesBufferController { pub(crate) fn start(self, handle: &Handle) { // TODO: should we do anything with JoinHandle(s) returned by start methods? - self.messsage_receiver.start(handle); + self.fragmented_messsage_receiver.start(handle); self.request_receiver.start(handle); } } diff --git a/clients/desktop/src/commands/init.rs b/clients/desktop/src/commands/init.rs index 2e57b4e8ea..196bb12bf3 100644 --- a/clients/desktop/src/commands/init.rs +++ b/clients/desktop/src/commands/init.rs @@ -43,14 +43,13 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> { .takes_value(true) ) .arg(Arg::with_name("directory") - .long("directory") - .help("Address of the directory server the client is getting topology from") - .takes_value(true), + .long("directory") + .help("Address of the directory server the client is getting topology from") + .takes_value(true), ) - .arg(Arg::with_name("socket-type") - .long("socket-type") - .help("Type of socket to use (TCP, WebSocket or None) in all subsequent runs") - .takes_value(true) + .arg(Arg::with_name("disable-socket") + .long("disable-socket") + .help("Whether to not start the websocket") ) .arg(Arg::with_name("port") .short("p") diff --git a/clients/desktop/src/commands/mod.rs b/clients/desktop/src/commands/mod.rs index fc1329caa5..f9bb333966 100644 --- a/clients/desktop/src/commands/mod.rs +++ b/clients/desktop/src/commands/mod.rs @@ -27,8 +27,8 @@ pub(crate) fn override_config(mut config: Config, matches: &ArgMatches) -> Confi config = config.with_gateway_id(gateway_id); } - if let Some(socket_type) = matches.value_of("socket-type") { - config = config.with_socket(SocketType::from_string(socket_type)); + if matches.is_present("disable-socket") { + config = config.with_socket(SocketType::None); } if let Some(port) = matches.value_of("port").map(|port| port.parse::()) { diff --git a/clients/desktop/src/commands/run.rs b/clients/desktop/src/commands/run.rs index 07048a071a..2138f2a7ad 100644 --- a/clients/desktop/src/commands/run.rs +++ b/clients/desktop/src/commands/run.rs @@ -34,19 +34,18 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> { .takes_value(true) ) .arg(Arg::with_name("directory") - .long("directory") - .help("Address of the directory server the client is getting topology from") - .takes_value(true), + .long("directory") + .help("Address of the directory server the client is getting topology from") + .takes_value(true), ) .arg(Arg::with_name("gateway") .long("gateway") .help("Id of the gateway we want to connect to. If overridden, it is user's responsibility to ensure prior registration happened") .takes_value(true) ) - .arg(Arg::with_name("socket-type") - .long("socket-type") - .help("Type of socket to use (TCP, WebSocket or None)") - .takes_value(true) + .arg(Arg::with_name("disable-socket") + .long("disable-socket") + .help("Whether to not start the websocket") ) .arg(Arg::with_name("port") .short("p") diff --git a/clients/desktop/src/config/mod.rs b/clients/desktop/src/config/mod.rs index 68e5d01b7d..51d52cb2f8 100644 --- a/clients/desktop/src/config/mod.rs +++ b/clients/desktop/src/config/mod.rs @@ -22,7 +22,7 @@ pub mod persistence; mod template; // 'CLIENT' -const DEFAULT_LISTENING_PORT: u16 = 9001; +const DEFAULT_LISTENING_PORT: u16 = 1977; // 'DEBUG' // where applicable, the below are defined in milliseconds @@ -41,7 +41,6 @@ const DEFAULT_NODE_SCORE_THRESHOLD: f64 = 0.0; #[derive(Debug, Deserialize, PartialEq, Serialize, Clone, Copy)] #[serde(deny_unknown_fields)] pub enum SocketType { - TCP, WebSocket, None, } @@ -51,7 +50,6 @@ impl SocketType { let mut upper = val.into(); upper.make_ascii_uppercase(); match upper.as_ref() { - "TCP" => SocketType::TCP, "WEBSOCKET" | "WS" => SocketType::WebSocket, _ => SocketType::None, } @@ -306,7 +304,7 @@ pub struct Socket { impl Default for Socket { fn default() -> Self { Socket { - socket_type: SocketType::None, + socket_type: SocketType::WebSocket, listening_port: DEFAULT_LISTENING_PORT, } } diff --git a/clients/desktop/src/config/template.rs b/clients/desktop/src/config/template.rs index f06acab0e1..0afa6c2b33 100644 --- a/clients/desktop/src/config/template.rs +++ b/clients/desktop/src/config/template.rs @@ -55,10 +55,10 @@ nym_root_directory = '{{ client.nym_root_directory }}' [socket] -# allowed values are 'TCP', 'WebSocket' or 'None' +# allowed values are 'WebSocket' or 'None' socket_type = '{{ socket.socket_type }}' -# if applicable (for the case of 'TCP' or 'WebSocket'), the port on which the client +# if applicable (for the case of 'WebSocket'), the port on which the client # will be listening for incoming requests listening_port = {{ socket.listening_port }} diff --git a/clients/desktop/src/lib.rs b/clients/desktop/src/lib.rs index faf849af6c..a1716e3ab7 100644 --- a/clients/desktop/src/lib.rs +++ b/clients/desktop/src/lib.rs @@ -15,4 +15,4 @@ pub mod built_info; pub mod client; pub mod config; -pub mod sockets; +pub mod websocket; diff --git a/clients/desktop/src/main.rs b/clients/desktop/src/main.rs index 7890cbeb48..949b36e067 100644 --- a/clients/desktop/src/main.rs +++ b/clients/desktop/src/main.rs @@ -18,7 +18,7 @@ pub mod built_info; pub mod client; mod commands; pub mod config; -pub mod sockets; +pub(crate) mod websocket; fn main() { dotenv::dotenv().ok(); diff --git a/clients/desktop/src/sockets/mod.rs b/clients/desktop/src/sockets/mod.rs deleted file mode 100644 index 55cf6b4308..0000000000 --- a/clients/desktop/src/sockets/mod.rs +++ /dev/null @@ -1,16 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -pub mod tcp; -pub mod websocket; diff --git a/clients/desktop/src/sockets/tcp.rs b/clients/desktop/src/sockets/tcp.rs deleted file mode 100644 index bed2e481c8..0000000000 --- a/clients/desktop/src/sockets/tcp.rs +++ /dev/null @@ -1,360 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::client::received_buffer::ReceivedBufferResponse; -use crate::client::topology_control::TopologyAccessor; -use crate::client::InputMessage; -use futures::channel::{mpsc, oneshot}; -use futures::future::FutureExt; -use futures::io::Error; -use futures::SinkExt; -use log::*; -use nymsphinx::chunking::split_and_prepare_payloads; -use nymsphinx::{Destination, DestinationAddressBytes}; -use std::convert::TryFrom; -use std::io; -use std::net::SocketAddr; -use tokio::prelude::*; -use tokio::runtime::Handle; -use tokio::task::JoinHandle; -use topology::NymTopology; - -const SEND_REQUEST_PREFIX: u8 = 1; -const FETCH_REQUEST_PREFIX: u8 = 2; -const GET_CLIENTS_REQUEST_PREFIX: u8 = 3; -const OWN_DETAILS_REQUEST_PREFIX: u8 = 4; - -#[derive(Debug)] -pub enum TCPSocketError { - FailedToStartSocketError, - UnknownSocketError, - IncompleteDataError, - UnknownRequestError, -} - -impl From for TCPSocketError { - fn from(err: Error) -> Self { - use TCPSocketError::*; - match err.kind() { - io::ErrorKind::ConnectionRefused => FailedToStartSocketError, - io::ErrorKind::ConnectionReset => FailedToStartSocketError, - io::ErrorKind::ConnectionAborted => FailedToStartSocketError, - io::ErrorKind::NotConnected => FailedToStartSocketError, - - io::ErrorKind::AddrInUse => FailedToStartSocketError, - io::ErrorKind::AddrNotAvailable => FailedToStartSocketError, - _ => UnknownSocketError, - } - } -} - -enum ClientRequest { - Send { - message: Vec, - recipient_address: DestinationAddressBytes, - }, - Fetch, - GetClients, - OwnDetails, -} - -impl TryFrom<&[u8]> for ClientRequest { - type Error = TCPSocketError; - - fn try_from(data: &[u8]) -> Result { - use TCPSocketError::*; - if data.is_empty() { - return Err(IncompleteDataError); - } - - match data[0] { - SEND_REQUEST_PREFIX => parse_send_request(data), - FETCH_REQUEST_PREFIX => Ok(ClientRequest::Fetch), - GET_CLIENTS_REQUEST_PREFIX => Ok(ClientRequest::GetClients), - OWN_DETAILS_REQUEST_PREFIX => Ok(ClientRequest::OwnDetails), - _ => Err(UnknownRequestError), - } - } -} - -fn parse_send_request(data: &[u8]) -> Result { - if data.len() < 1 + 32 + 1 { - // make sure it has the prefix, destination and at least single byte of data - return Err(TCPSocketError::IncompleteDataError); - } - - let mut recipient_address = [0u8; 32]; - recipient_address.copy_from_slice(&data[1..33]); - - let message = data[33..].to_vec(); - - Ok(ClientRequest::Send { - message, - recipient_address: DestinationAddressBytes::from_bytes(recipient_address), - }) -} - -impl ClientRequest { - async fn handle_send( - msg: Vec, - recipient_address: DestinationAddressBytes, - input_tx: mpsc::UnboundedSender, - ) -> ServerResponse { - trace!("sending to: {:?}, msg: {:?}", recipient_address, msg); - let dummy_surb = [0; 16]; - - // in case the message is too long and needs to be split into multiple packets: - let split_message = split_and_prepare_payloads(&msg); - for message_fragment in split_message { - let input_msg = InputMessage( - Destination::new(recipient_address.clone(), dummy_surb), - message_fragment, - ); - input_tx.unbounded_send(input_msg).unwrap(); - } - - ServerResponse::Send - } - - async fn handle_fetch( - mut msg_query: mpsc::UnboundedSender, - ) -> ServerResponse { - trace!("handle_fetch called"); - let (res_tx, res_rx) = oneshot::channel(); - if msg_query.send(res_tx).await.is_err() { - let e = "Nym-client TCP socket failed to receive messages".to_string(); - error!("{}", e); - return ServerResponse::Error { message: e }; - } - - let messages = match res_rx.map(|msg| msg).await { - Ok(messages) => messages, - Err(e) => { - warn!("Failed to fetch client messages - {:?}", e); - return ServerResponse::Error { - message: "Server failed to receive messages".to_string(), - }; - } - }; - - trace!("fetched {} messages", messages.len()); - ServerResponse::Fetch { messages } - } - - async fn handle_get_clients( - mut topology_accessor: TopologyAccessor, - ) -> ServerResponse { - match topology_accessor.get_current_topology_clone().await { - Some(topology) => { - let clients = topology - .providers() - .iter() - .flat_map(|provider| provider.registered_clients.iter()) - .filter_map(|client| bs58::decode(&client.pub_key).into_vec().ok()) - .collect(); - ServerResponse::GetClients { clients } - } - None => ServerResponse::Error { - message: "Invalid network topology".to_string(), - }, - } - } - - async fn handle_own_details(self_address_bytes: DestinationAddressBytes) -> ServerResponse { - ServerResponse::OwnDetails { - address: self_address_bytes.to_bytes().to_vec(), - } - } -} - -enum ServerResponse { - Send, - Fetch { messages: Vec> }, - GetClients { clients: Vec> }, - OwnDetails { address: Vec }, - Error { message: String }, -} - -impl Into> for ServerResponse { - fn into(self) -> Vec { - match self { - ServerResponse::Send => b"ok".to_vec(), - ServerResponse::Fetch { messages } => encode_fetched_messages(messages), - ServerResponse::GetClients { clients } => encode_list_of_clients(clients), - ServerResponse::OwnDetails { address } => address, - ServerResponse::Error { message } => message.as_bytes().to_vec(), - } - } -} - -// num_msgs || len1 || len2 || ... || msg1 || msg2 || ... -fn encode_fetched_messages(messages: Vec>) -> Vec { - // for reciprocal of this look into sfw-provider-requests::responses::PullResponse::from_bytes() - - let num_msgs = messages.len() as u16; - let msgs_lens: Vec = messages.iter().map(|msg| msg.len() as u16).collect(); - - num_msgs - .to_be_bytes() - .to_vec() - .into_iter() - .chain( - msgs_lens - .into_iter() - .flat_map(|len| len.to_be_bytes().to_vec().into_iter()), - ) - .chain(messages.iter().flat_map(|msg| msg.clone().into_iter())) - .collect() -} - -fn encode_list_of_clients(clients: Vec>) -> Vec { - debug!("client: {:?}", clients); - // we can just concat all client since all of them got to be 32 bytes long - // (if not, then we have bigger problem somewhere up the line) - - // converts [[1,2,3],[4,5,6],...] into [1,2,3,4,5,6,...] - clients.into_iter().flatten().collect() -} - -impl ServerResponse { - fn new_error(message: String) -> ServerResponse { - ServerResponse::Error { message } - } -} - -async fn handle_connection( - data: &[u8], - request_handling_data: RequestHandlingData, -) -> Result { - let request = ClientRequest::try_from(data)?; - let response = match request { - ClientRequest::Send { - message, - recipient_address, - } => { - ClientRequest::handle_send(message, recipient_address, request_handling_data.msg_input) - .await - } - ClientRequest::Fetch => ClientRequest::handle_fetch(request_handling_data.msg_query).await, - ClientRequest::GetClients => { - ClientRequest::handle_get_clients(request_handling_data.topology_accessor).await - } - ClientRequest::OwnDetails => { - ClientRequest::handle_own_details(request_handling_data.self_address).await - } - }; - - Ok(response) -} - -struct RequestHandlingData { - msg_input: mpsc::UnboundedSender, - msg_query: mpsc::UnboundedSender, - self_address: DestinationAddressBytes, - topology_accessor: TopologyAccessor, -} - -async fn accept_connection( - mut socket: tokio::net::TcpStream, - msg_input: mpsc::UnboundedSender, - msg_query: mpsc::UnboundedSender, - self_address: DestinationAddressBytes, - topology_accessor: TopologyAccessor, -) { - let address = socket - .peer_addr() - .expect("connected streams should have a peer address"); - debug!("Peer address: {}", address); - - let mut buf = [0u8; 2048]; - - // In a loop, read data from the socket and write the data back. - loop { - // TODO: shutdowns? - - let response = match socket.read(&mut buf).await { - // socket closed - Ok(n) if n == 0 => { - trace!("Remote connection closed."); - return; - } - Ok(n) => { - let request_handling_data = RequestHandlingData { - topology_accessor: topology_accessor.clone(), - msg_input: msg_input.clone(), - msg_query: msg_query.clone(), - self_address: self_address.clone(), - }; - match handle_connection(&buf[..n], request_handling_data).await { - Ok(res) => res, - Err(e) => ServerResponse::new_error(format!("{:?}", e)), - } - } - Err(e) => { - warn!("failed to read from socket; err = {:?}", e); - return; - } - }; - - let response_vec: Vec = response.into(); - if let Err(e) = socket.write_all(&response_vec).await { - warn!("failed to write reply to socket; err = {:?}", e); - return; - } - } -} - -pub(crate) async fn run_tcpsocket( - listening_port: u16, - message_tx: mpsc::UnboundedSender, - received_messages_query_tx: mpsc::UnboundedSender, - self_address: DestinationAddressBytes, - topology_accessor: TopologyAccessor, -) { - let address = SocketAddr::new("127.0.0.1".parse().unwrap(), listening_port); - info!("Starting tcp socket listener at {}", address.to_string()); - let mut listener = tokio::net::TcpListener::bind(address).await.unwrap(); - - while let Ok((stream, _)) = listener.accept().await { - // it's fine to be cloning the channel on all new connection, because in principle - // this server should only EVER have a single client connected - tokio::spawn(accept_connection( - stream, - message_tx.clone(), - received_messages_query_tx.clone(), - self_address.clone(), - topology_accessor.clone(), - )); - } -} - -pub(crate) fn start_tcpsocket( - handle: &Handle, - listening_port: u16, - message_tx: mpsc::UnboundedSender, - received_messages_query_tx: mpsc::UnboundedSender, - self_address: DestinationAddressBytes, - topology_accessor: TopologyAccessor, -) -> JoinHandle<()> { - handle.spawn(async move { - run_tcpsocket( - listening_port, - message_tx, - received_messages_query_tx, - self_address, - topology_accessor, - ) - .await; - }) -} diff --git a/clients/desktop/src/sockets/websocket/connection.rs b/clients/desktop/src/sockets/websocket/connection.rs deleted file mode 100644 index 43f4be2125..0000000000 --- a/clients/desktop/src/sockets/websocket/connection.rs +++ /dev/null @@ -1,282 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::client::received_buffer::ReceivedBufferRequestSender; -use crate::client::topology_control::TopologyAccessor; -use crate::client::{InputMessage, InputMessageSender}; -use crate::sockets::websocket::types::{ClientRequest, ServerResponse}; -use futures::channel::oneshot; -use futures::{SinkExt, StreamExt}; -use log::*; -use nymsphinx::chunking::split_and_prepare_payloads; -use nymsphinx::{Destination, DestinationAddressBytes}; -use std::convert::TryFrom; -use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode; -use tokio_tungstenite::tungstenite::protocol::{CloseFrame, Message}; -use tokio_tungstenite::WebSocketStream; -use topology::NymTopology; - -#[derive(Clone)] -pub(crate) struct ConnectionData { - msg_input: InputMessageSender, - msg_query: ReceivedBufferRequestSender, - self_address: DestinationAddressBytes, - topology_accessor: TopologyAccessor, -} - -impl ConnectionData { - pub(crate) fn new( - msg_input: InputMessageSender, - msg_query: ReceivedBufferRequestSender, - self_address: DestinationAddressBytes, - topology_accessor: TopologyAccessor, - ) -> Self { - ConnectionData { - msg_input, - msg_query, - self_address, - topology_accessor, - } - } -} - -pub(crate) struct Connection { - ws_stream: WebSocketStream, - - msg_input: InputMessageSender, - msg_query: ReceivedBufferRequestSender, - - self_address: DestinationAddressBytes, - topology_accessor: TopologyAccessor, -} - -impl Connection { - pub(crate) async fn try_accept( - raw_stream: tokio::net::TcpStream, - connection_data: ConnectionData, - ) -> Option { - // perform ws handshake - let ws_stream = match tokio_tungstenite::accept_async(raw_stream).await { - Ok(ws_stream) => ws_stream, - Err(e) => { - error!("Error during the websocket handshake occurred - {}", e); - return None; - } - }; - - Some(Connection { - ws_stream, - msg_input: connection_data.msg_input, - msg_query: connection_data.msg_query, - self_address: connection_data.self_address, - topology_accessor: connection_data.topology_accessor, - }) - } - - fn handle_text_send(&self, msg: String, recipient_address: String) -> ServerResponse { - let message_bytes = msg.into_bytes(); - - let address = match DestinationAddressBytes::try_from_base58_string(recipient_address) { - Ok(address) => address, - Err(e) => { - trace!("failed to parse received DestinationAddress: {:?}", e); - return ServerResponse::Error { - message: "malformed destination address".to_string(), - }; - } - }; - let dummy_surb = [0; 16]; - - // in case the message is too long and needs to be split into multiple packets: - let split_message = split_and_prepare_payloads(&message_bytes); - for message_fragment in split_message { - let input_msg = InputMessage( - Destination::new(address.clone(), dummy_surb), - message_fragment, - ); - self.msg_input.unbounded_send(input_msg).unwrap(); - } - - ServerResponse::Send - } - - async fn handle_text_fetch(&self) -> ServerResponse { - // send the request to the buffer controller - let (res_tx, res_rx) = oneshot::channel(); - self.msg_query.unbounded_send(res_tx).unwrap(); - let messages_bytes = res_rx.await.unwrap(); - - let messages = messages_bytes - .into_iter() - .map(|message| { - std::str::from_utf8(&message) - .unwrap_or_else(|_| { - error!("Invalid UTF-8 sequence in response message"); - "" - }) - .into() - }) - .collect(); - - ServerResponse::Fetch { messages } - } - - async fn handle_text_get_clients(&mut self) -> ServerResponse { - match self.topology_accessor.get_all_clients().await { - Some(clients) => { - let client_keys = clients.into_iter().map(|client| client.pub_key).collect(); - ServerResponse::GetClients { - clients: client_keys, - } - } - None => ServerResponse::Error { - message: "Invalid network topology".to_string(), - }, - } - } - - fn handle_text_own_details(&self) -> ServerResponse { - ServerResponse::OwnDetails { - address: self.self_address.to_base58_string(), - } - } - - async fn handle_text_message(&mut self, msg: String) -> Message { - debug!("Handling text message request"); - trace!("Content: {:?}", msg.clone()); - - match ClientRequest::try_from(msg) { - Err(e) => ServerResponse::Error { - message: format!("received invalid request. err: {:?}", e), - } - .into(), - Ok(req) => match req { - ClientRequest::Send { - message, - recipient_address, - } => self.handle_text_send(message, recipient_address), - ClientRequest::Fetch => self.handle_text_fetch().await, - ClientRequest::GetClients => self.handle_text_get_clients().await, - ClientRequest::OwnDetails => self.handle_text_own_details(), - } - .into(), - } - } - - // 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_request(&mut self, raw_request: Message) -> Message { - match raw_request { - Message::Text(text_message) => self.handle_text_message(text_message).await, - 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, - } - } - - pub(crate) async fn start_handling(&mut self) { - while let Some(raw_message) = self.ws_stream.next().await { - let raw_message = match raw_message { - Ok(msg) => msg, - Err(err) => { - error!("failed to obtain message from websocket stream! stopping connection handler: {}", err); - return; - } - }; - - let response = self.handle_request(raw_message).await; - let is_close = response.is_close(); - if let Err(err) = self.ws_stream.send(response).await { - warn!( - "Failed to send message over websocket: {}. Assuming the connection is dead.", - 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 is_close { - info!("Closing the websocket connection"); - return; - } - } - } -} diff --git a/clients/desktop/src/sockets/websocket/listener.rs b/clients/desktop/src/sockets/websocket/listener.rs deleted file mode 100644 index 4edd656bcb..0000000000 --- a/clients/desktop/src/sockets/websocket/listener.rs +++ /dev/null @@ -1,63 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::client::received_buffer::ReceivedBufferRequestSender; -use crate::client::topology_control::TopologyAccessor; -use crate::client::InputMessageSender; -use crate::sockets::websocket::connection::{Connection, ConnectionData}; -use log::*; -use nymsphinx::DestinationAddressBytes; -use std::net::SocketAddr; -use tokio::runtime::Handle; -use tokio::task::JoinHandle; -use topology::NymTopology; - -async fn process_socket_connection( - stream: tokio::net::TcpStream, - connection_data: ConnectionData, -) { - match Connection::try_accept(stream, connection_data).await { - None => warn!("Failed to establish websocket connection"), - Some(mut conn) => conn.start_handling().await, - } -} - -pub(crate) fn run( - handle: &Handle, - port: u16, - msg_input: InputMessageSender, - msg_query: ReceivedBufferRequestSender, - self_address: DestinationAddressBytes, - topology_accessor: TopologyAccessor, -) -> JoinHandle<()> { - let handle_clone = handle.clone(); - handle.spawn(async move { - let address = SocketAddr::new("127.0.0.1".parse().unwrap(), port); - info!("Starting websocket listener at {:?}", address); - let mut listener = tokio::net::TcpListener::bind(address).await.unwrap(); - let connection_data = - ConnectionData::new(msg_input, msg_query, self_address, topology_accessor); - - // in theory there should only ever be a single connection made to the listener - // but it's not significantly more difficult to allow more of them if needed - loop { - let (stream, _) = listener.accept().await.unwrap(); - { - let connection_data = connection_data.clone(); - handle_clone - .spawn(async move { process_socket_connection(stream, connection_data).await }); - } - } - }) -} diff --git a/clients/desktop/src/sockets/websocket/types.rs b/clients/desktop/src/sockets/websocket/types.rs deleted file mode 100644 index df34404fe3..0000000000 --- a/clients/desktop/src/sockets/websocket/types.rs +++ /dev/null @@ -1,56 +0,0 @@ -// Copyright 2020 Nym Technologies SA -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use serde::{Deserialize, Serialize}; -use std::convert::TryFrom; -use tokio_tungstenite::tungstenite::protocol::Message; - -#[derive(Serialize, Deserialize, Debug)] -#[serde(tag = "type", rename_all = "camelCase")] -pub(crate) enum ClientRequest { - Send { - message: String, - recipient_address: String, - }, - Fetch, - GetClients, - OwnDetails, -} - -impl TryFrom for ClientRequest { - type Error = serde_json::Error; - - fn try_from(msg: String) -> Result { - serde_json::from_str(&msg) - } -} - -#[derive(Serialize, Deserialize, Debug)] -#[serde(tag = "type", rename_all = "camelCase")] -pub(crate) enum ServerResponse { - Send, - Fetch { messages: Vec }, - GetClients { clients: Vec }, - OwnDetails { address: String }, - Error { message: String }, -} - -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) - } -} diff --git a/clients/desktop/src/websocket/handler.rs b/clients/desktop/src/websocket/handler.rs new file mode 100644 index 0000000000..1eb2999721 --- /dev/null +++ b/clients/desktop/src/websocket/handler.rs @@ -0,0 +1,334 @@ +// Copyright 2020 Nym Technologies SA +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::types::{BinaryClientRequest, ClientRequest, ServerResponse}; +use crate::client::{ + received_buffer::{ + ReceivedBufferMessage, ReceivedBufferRequestSender, ReconstructeredMessagesReceiver, + }, + topology_control::TopologyAccessor, + InputMessage, InputMessageSender, +}; +use futures::channel::mpsc; +use futures::{SinkExt, StreamExt}; +use log::*; +use nymsphinx::chunking::split_and_prepare_payloads; +use nymsphinx::{Destination, DestinationAddressBytes}; +use std::convert::TryFrom; +use tokio::net::TcpStream; +use tokio_tungstenite::{ + accept_async, + tungstenite::{protocol::Message, Error as WsError}, + WebSocketStream, +}; +use topology::NymTopology; + +enum ReceivedResponseType { + Binary, + Text, +} + +impl Default for ReceivedResponseType { + fn default() -> Self { + ReceivedResponseType::Binary + } +} + +pub(crate) struct Handler { + msg_input: InputMessageSender, + buffer_requester: ReceivedBufferRequestSender, + self_address: DestinationAddressBytes, + topology_accessor: TopologyAccessor, + socket: Option>, + received_response_type: ReceivedResponseType, +} + +// clone is used to use handler on a new connection, which initially is `None` +impl Clone for Handler { + fn clone(&self) -> Self { + Handler { + msg_input: self.msg_input.clone(), + buffer_requester: self.buffer_requester.clone(), + self_address: self.self_address.clone(), + topology_accessor: self.topology_accessor.clone(), + socket: None, + received_response_type: Default::default(), + } + } +} + +impl Drop for Handler { + fn drop(&mut self) { + self.buffer_requester + .unbounded_send(ReceivedBufferMessage::ReceiverDisconnect) + .expect("the buffer request failed!") + } +} + +impl Handler { + pub(crate) fn new( + msg_input: InputMessageSender, + buffer_requester: ReceivedBufferRequestSender, + self_address: DestinationAddressBytes, + topology_accessor: TopologyAccessor, + ) -> Self { + Handler { + msg_input, + buffer_requester, + self_address, + topology_accessor, + socket: None, + received_response_type: Default::default(), + } + } + + fn handle_text_send(&mut self, msg: String, recipient_address: String) -> ServerResponse { + let message_bytes = msg.into_bytes(); + + let address = match DestinationAddressBytes::try_from_base58_string(recipient_address) { + Ok(address) => address, + Err(e) => { + trace!("failed to parse received DestinationAddress: {:?}", e); + return ServerResponse::new_error("malformed destination address"); + } + }; + + // in case the message is too long and needs to be split into multiple packets: + let split_message = split_and_prepare_payloads(&message_bytes); + for message_fragment in split_message { + let input_msg = InputMessage( + Destination::new(address.clone(), Default::default()), + message_fragment, + ); + self.msg_input.unbounded_send(input_msg).unwrap(); + } + + self.received_response_type = ReceivedResponseType::Text; + + ServerResponse::Send + } + + async fn handle_text_get_clients(&mut self) -> ServerResponse { + match self.topology_accessor.get_all_clients().await { + Some(clients) => { + let client_keys = clients.into_iter().map(|client| client.pub_key).collect(); + ServerResponse::GetClients { + clients: client_keys, + } + } + None => ServerResponse::new_error("invalid network topology"), + } + } + + fn handle_text_self_address(&self) -> ServerResponse { + ServerResponse::SelfAddress { + address: self.self_address.to_base58_string(), + } + } + + async fn handle_text_message(&mut self, msg: String) -> Message { + debug!("Handling text message request"); + trace!("Content: {:?}", msg.clone()); + + match ClientRequest::try_from(msg) { + Err(e) => ServerResponse::Error { + message: format!("received invalid request. err: {:?}", e), + } + .into(), + Ok(req) => match req { + ClientRequest::Send { message, recipient } => { + self.handle_text_send(message, recipient) + } + ClientRequest::GetClients => self.handle_text_get_clients().await, + ClientRequest::SelfAddress => self.handle_text_self_address(), + } + .into(), + } + } + + async fn handle_binary_send( + &mut self, + address: DestinationAddressBytes, + data: Vec, + ) -> ServerResponse { + // in case the message is too long and needs to be split into multiple packets: + let split_message = split_and_prepare_payloads(&data); + for message_fragment in split_message { + let input_msg = InputMessage( + Destination::new(address.clone(), Default::default()), + message_fragment, + ); + self.msg_input.unbounded_send(input_msg).unwrap(); + } + + self.received_response_type = ReceivedResponseType::Binary; + ServerResponse::Send + } + + // if it's binary we assume it's a sphinx packet formatted the same way as we'd have sent + // it to the gateway + async fn handle_binary_message(&mut self, msg: Vec) -> Message { + debug!("Handling binary message request"); + + self.received_response_type = ReceivedResponseType::Binary; + // make sure it is correctly formatted + let binary_request = BinaryClientRequest::try_from_bytes(&msg); + if binary_request.is_none() { + return ServerResponse::new_error("invalid binary request").into(); + } + match binary_request.unwrap() { + BinaryClientRequest::Send { + recipient_address, + data, + } => self.handle_binary_send(recipient_address, data).await, + } + .into() + } + + async fn handle_request(&mut self, raw_request: Message) -> Option { + // apparently tungstenite auto-handles ping/pong/close messages so for now let's ignore + // them and let's test that claim. If that's not the case, just copy code from + // old version of this file. + match raw_request { + Message::Text(text_message) => Some(self.handle_text_message(text_message).await), + Message::Binary(binary_message) => { + Some(self.handle_binary_message(binary_message).await) + } + _ => None, + } + } + + async fn push_websocket_received_plaintexts( + &mut self, + messages_bytes: Vec>, + ) -> Result<(), WsError> { + let response_messages: Vec<_> = match self.received_response_type { + ReceivedResponseType::Binary => messages_bytes + .into_iter() + .map(|msg| Ok(Message::Binary(msg))) + .collect(), + ReceivedResponseType::Text => { + let mut decoded_messages = Vec::new(); + // either all succeed or all fall back + let mut did_fail = false; + for message in messages_bytes.iter() { + match std::str::from_utf8(message) { + Ok(msg) => decoded_messages.push(msg), + Err(err) => { + did_fail = true; + warn!("Invalid UTF-8 sequence in response message - {:?}", err); + break; + } + } + } + if did_fail { + messages_bytes + .into_iter() + .map(|msg| Ok(Message::Binary(msg))) + .collect() + } else { + decoded_messages + .into_iter() + .map(|msg| Ok(Message::Text(msg.to_string()))) + .collect() + } + } + }; + + let mut send_stream = futures::stream::iter(response_messages); + self.socket + .as_mut() + .unwrap() + .send_all(&mut send_stream) + .await + } + + async fn send_websocket_response(&mut self, msg: Message) -> Result<(), WsError> { + match self.socket { + // TODO: more closely investigate difference between `Sink::send` and `Sink::send_all` + // it got something to do with batching and flushing - it might be important if it + // turns out somehow we've got a bottleneck here + Some(ref mut ws_stream) => ws_stream.send(msg).await, + _ => panic!("impossible state - websocket handshake was somehow reverted"), + } + } + + async fn next_websocket_request(&mut self) -> Option> { + match self.socket { + Some(ref mut ws_stream) => ws_stream.next().await, + None => None, + } + } + + async fn listen_for_requests(&mut self, mut msg_receiver: ReconstructeredMessagesReceiver) { + loop { + tokio::select! { + socket_msg = self.next_websocket_request() => { + if socket_msg.is_none() { + break; + } + let socket_msg = match socket_msg.unwrap() { + Ok(socket_msg) => socket_msg, + Err(err) => { + error!("failed to obtain message from websocket stream! stopping connection handler: {}", err); + break; + } + }; + + if socket_msg.is_close() { + break; + } + + if let Some(response) = self.handle_request(socket_msg).await { + if let Err(err) = self.send_websocket_response(response).await { + warn!( + "Failed to send message over websocket: {}. Assuming the connection is dead.", + err + ); + break; + } + } + } + mix_messages = msg_receiver.next() => { + let mix_messages = mix_messages.expect( + "mix messages sender was unexpectedly closed! this shouldn't have ever happened!", + ); + if let Err(e) = self.push_websocket_received_plaintexts(mix_messages).await { + warn!("failed to send sphinx packets back to the client - {:?}, assuming the connection is dead", e); + break; + } + } + } + } + } + + // consume self to make sure `drop` is called after this is done + pub(crate) async fn handle_connection(mut self, socket: TcpStream) { + let ws_stream = accept_async(socket) + .await + .expect("error while performing the websocket handshake"); + self.socket = Some(ws_stream); + + let (reconstructed_sender, reconstructed_receiver) = mpsc::unbounded(); + + // tell the buffer to start sending stuff to us + self.buffer_requester + .unbounded_send(ReceivedBufferMessage::ReceiverAnnounce( + reconstructed_sender, + )) + .expect("the buffer request failed!"); + + self.listen_for_requests(reconstructed_receiver).await; + } +} diff --git a/clients/desktop/src/websocket/listener.rs b/clients/desktop/src/websocket/listener.rs new file mode 100644 index 0000000000..b47c355feb --- /dev/null +++ b/clients/desktop/src/websocket/listener.rs @@ -0,0 +1,115 @@ +// Copyright 2020 Nym Technologies SA +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::handler::Handler; +use log::*; +use std::{ + net::{Shutdown, SocketAddr}, + sync::Arc, +}; +use tokio::runtime; +use tokio::{sync::Notify, task::JoinHandle}; +use topology::NymTopology; + +enum State { + Connected, + AwaitingConnection, +} + +impl State { + fn is_connected(&self) -> bool { + match self { + State::Connected => true, + _ => false, + } + } +} + +pub(crate) struct Listener { + address: SocketAddr, + state: State, +} + +impl Listener { + pub(crate) fn new(port: u16) -> Self { + Listener { + // unless we find compelling reason not to, just listen on local only + address: SocketAddr::new("127.0.0.1".parse().unwrap(), port), + state: State::AwaitingConnection, + } + } + + pub(crate) async fn run(&mut self, handler: Handler) { + let mut tcp_listener = tokio::net::TcpListener::bind(self.address) + .await + .expect("Failed to start websocket listener"); + + let notify = Arc::new(Notify::new()); + + loop { + tokio::select! { + _ = notify.notified() => { + // our connection terminated - we are open to a new one now! + self.state = State::AwaitingConnection; + } + new_conn = tcp_listener.accept() => { + match new_conn { + Ok((socket, remote_addr)) => { + debug!("Received connection from {:?}", remote_addr); + if self.state.is_connected() { + warn!("tried to duplicate!"); + // if we've already got a connection, don't allow another one + debug!("but there was already a connection present!"); + // while we only ever want to accept a single connection, we don't want + // to leave clients hanging (and also allow for reconnection if it somehow + // was dropped) + match socket.shutdown(Shutdown::Both) { + Ok(_) => trace!( + "closed the connection between attempting websocket handshake" + ), + Err(e) => warn!("failed to cleanly close the connection - {:?}", e), + }; + } else { + // even though we're spawning a new task with the handler here, we will only ever spawn a single one. + // it's done so that any new connections to this listener could be rejected rather than left + // hanging because the executor doesn't come back here + let notify_clone = Arc::clone(¬ify); + let fresh_handler = handler.clone(); + tokio::spawn(async move { + fresh_handler.handle_connection(socket).await; + notify_clone.notify(); + }); + self.state = State::Connected; + } + } + Err(e) => warn!("failed to get client: {:?}", e), + } + } + } + } + } + + pub(crate) fn start( + mut self, + rt_handle: &runtime::Handle, + handler: Handler, + ) -> JoinHandle<()> { + info!( + "The websocket listener will try to run on {:?}", + self.address.to_string() + ); + + rt_handle.spawn(async move { self.run(handler).await }) + } +} diff --git a/clients/desktop/src/sockets/websocket/mod.rs b/clients/desktop/src/websocket/mod.rs similarity index 79% rename from clients/desktop/src/sockets/websocket/mod.rs rename to clients/desktop/src/websocket/mod.rs index d2b44ebfeb..a5af5e8121 100644 --- a/clients/desktop/src/sockets/websocket/mod.rs +++ b/clients/desktop/src/websocket/mod.rs @@ -12,6 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub(crate) mod connection; +pub(crate) mod handler; pub(crate) mod listener; pub(crate) mod types; + +pub(crate) use handler::Handler; +pub(crate) use listener::Listener; + +pub use types::{BinaryClientRequest, ClientRequest, ServerResponse}; diff --git a/clients/desktop/src/websocket/types.rs b/clients/desktop/src/websocket/types.rs new file mode 100644 index 0000000000..4e338ae2b5 --- /dev/null +++ b/clients/desktop/src/websocket/types.rs @@ -0,0 +1,119 @@ +// Copyright 2020 Nym Technologies SA +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use nymsphinx::{DestinationAddressBytes, DESTINATION_ADDRESS_LENGTH}; +use serde::{Deserialize, Serialize}; +use std::convert::TryFrom; +use tokio_tungstenite::tungstenite::protocol::Message; + +#[derive(Serialize, Deserialize, Debug)] +#[serde(tag = "type", rename_all = "camelCase")] +pub enum ClientRequest { + Send { message: String, recipient: String }, + GetClients, + SelfAddress, +} + +impl TryFrom for ClientRequest { + type Error = serde_json::Error; + + fn try_from(msg: String) -> Result { + serde_json::from_str(&msg) + } +} + +impl Into for ClientRequest { + fn into(self) -> Message { + let str_req = serde_json::to_string(&self).unwrap(); + Message::Text(str_req) + } +} + +pub enum BinaryClientRequest { + Send { + recipient_address: DestinationAddressBytes, + data: Vec, + }, +} + +impl BinaryClientRequest { + // TODO: perhaps do it the proper way and introduce an error type + pub fn try_from_bytes(req: &[u8]) -> Option { + if req.len() < DESTINATION_ADDRESS_LENGTH { + return None; + } + let mut destination_bytes = [0u8; DESTINATION_ADDRESS_LENGTH]; + destination_bytes.copy_from_slice(&req[..DESTINATION_ADDRESS_LENGTH]); + let address = DestinationAddressBytes::from_bytes(destination_bytes); + Some(BinaryClientRequest::Send { + recipient_address: address, + data: req[DESTINATION_ADDRESS_LENGTH..].to_vec(), + }) + } + + pub fn into_bytes(self) -> Vec { + match self { + Self::Send { + recipient_address, + data, + } => recipient_address + .to_bytes() + .iter() + .cloned() + .chain(data.into_iter()) + .collect(), + } + } +} + +impl Into for BinaryClientRequest { + fn into(self) -> Message { + Message::Binary(self.into_bytes()) + } +} + +#[derive(Serialize, Deserialize, Debug)] +#[serde(tag = "type", rename_all = "camelCase")] +pub enum ServerResponse { + Send, + Received { messages: Vec }, + GetClients { clients: Vec }, + SelfAddress { address: String }, + Error { message: String }, +} + +impl ServerResponse { + pub fn new_error>(msg: S) -> Self { + ServerResponse::Error { + message: msg.into(), + } + } +} + +impl TryFrom for ServerResponse { + type Error = serde_json::Error; + + fn try_from(msg: String) -> Result>::Error> { + serde_json::from_str(&msg) + } +} + +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) + } +}