GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'develop' into feature/new_websocket_implementation

Browse Source
This commit is contained in:
Jedrzej Stuczynski
2019-12-17 15:57:59 +00:00
parent c02a44fbce 5cde753e6c
commit 2f871ea320
14 changed files with 148 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/clients/directory/presence.rs
+5 -5
View File
@@ -1,6 +1,6 @@
use serde::{Deserialize, Serialize};
#[derive(Deserialize, Serialize, Clone)]
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CocoPresence {
pub host: String,
@@ -8,7 +8,7 @@ pub struct CocoPresence {
pub last_seen: i64,
}
#[derive(Deserialize, Serialize, Clone)]
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct MixNodePresence {
pub host: String,
@@ -17,7 +17,7 @@ pub struct MixNodePresence {
pub last_seen: u64,
}
#[derive(Deserialize, Serialize, Clone)]
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct MixProviderPresence {
pub host: String,
@@ -25,14 +25,14 @@ pub struct MixProviderPresence {
pub registered_clients: Vec<MixProviderClient>,
}
#[derive(Deserialize, Serialize, Clone)]
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct MixProviderClient {
pub pub_key: String,
}
// Topology shows us the current state of the overall Nym network
#[derive(Deserialize, Serialize, Clone)]
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Topology {
pub coco_nodes: Vec<CocoPresence>,
src/clients/mix.rs
+1 -2
View File
@@ -1,6 +1,5 @@
use sphinx::route::{Node as MixNode, NodeAddressBytes};
use sphinx::route::NodeAddressBytes;
use sphinx::SphinxPacket;
use std::net::SocketAddr;
use std::net::{Ipv4Addr, SocketAddrV4};
use tokio::prelude::*;
src/clients/mod.rs
+36 -12
View File
@@ -1,17 +1,19 @@
use crate::clients::directory::presence::Topology;
use crate::clients::mix::MixClient;
use crate::clients::provider::ProviderClient;
use crate::sockets::ws;
use crate::utils;
use crate::utils::topology::get_topology;
use futures::channel::mpsc;
use futures::future::join5;
use futures::select;
use futures::{SinkExt, StreamExt};
use sphinx::route::{Destination, DestinationAddressBytes, NodeAddressBytes};
use sphinx::SphinxPacket;
use std::net::SocketAddr;
use std::net::SocketAddrV4;
use std::time::Duration;
use tokio::runtime::Runtime;
use futures::future::join5;
use crate::sockets::ws;
use std::net::SocketAddr;
pub mod directory;
pub mod mix;
@@ -93,7 +95,8 @@ impl NymClient {
let cover_message =
utils::sphinx::loop_cover_message(our_info.address, our_info.identifier, &topology);
tx.send(MixMessage(cover_message.0, cover_message.1))
.await.unwrap();
.await
.unwrap();
}
}
@@ -119,20 +122,24 @@ impl NymClient {
let cover_message = utils::sphinx::loop_cover_message(our_info.address, our_info.identifier, &topology);
mix_tx.send(MixMessage(cover_message.0, cover_message.1)).await.unwrap();
}
}
;
};
let delay_duration = Duration::from_secs_f64(MESSAGE_SENDING_AVERAGE_DELAY);
tokio::time::delay_for(delay_duration).await;
}
}
// TODO: proper return type
async fn start_provider_polling() {
async fn start_provider_polling(provider_address: SocketAddrV4) {
let provider_client = ProviderClient::new();
loop {
println!("[FETCH MSG] - Polling provider...");
let delay_duration = Duration::from_secs_f64(FETCH_MESSAGES_DELAY);
tokio::time::delay_for(delay_duration).await;
provider_client
.retrieve_messages(provider_address)
.await
.unwrap();
}
}
@@ -141,8 +148,14 @@ impl NymClient {
let (mix_tx, mix_rx) = mpsc::unbounded();
let mut rt = Runtime::new()?;
let topology = get_topology(self.is_local);
let provider_address: SocketAddrV4 = topology
.mix_provider_nodes
.first()
.unwrap()
.host
.parse()
.unwrap();
let mix_traffic_future = rt.spawn(MixTrafficController::run(mix_rx));
let loop_cover_traffic_future = rt.spawn(NymClient::start_loop_cover_traffic_stream(
@@ -158,13 +171,24 @@ impl NymClient {
topology.clone(),
));
let provider_polling_future = rt.spawn(NymClient::start_provider_polling());
let provider_polling_future = rt.spawn(NymClient::start_provider_polling(provider_address));
let websocket_future = rt.spawn(ws::start_websocket(socket_address, self.input_tx));
rt.block_on(async {
let future_results = join5(mix_traffic_future, loop_cover_traffic_future, out_queue_control_future, provider_polling_future, websocket_future).await;
let future_results = join5(
mix_traffic_future,
loop_cover_traffic_future,
out_queue_control_future,
provider_polling_future,
websocket_future,
)
.await;
assert!(
future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok() && future_results.3.is_ok() && future_results.4.is_ok()
future_results.0.is_ok()
&& future_results.1.is_ok()
&& future_results.2.is_ok()
&& future_results.3.is_ok()
&& future_results.4.is_ok()
);
});
src/clients/provider.rs
+9 -2
View File
@@ -1,6 +1,7 @@
use sfw_provider_requests::requests::{ProviderRequest, PullRequest};
use sfw_provider_requests::responses::{ProviderResponse, PullResponse};
use std::net::Shutdown;
use std::net::SocketAddrV4;
use tokio::prelude::*;
use tokio::time::Duration;
@@ -13,13 +14,19 @@ impl ProviderClient {
pub async fn retrieve_messages(
&self,
// provider: &MixNode,
provider: SocketAddrV4,
) -> Result<(), Box<dyn std::error::Error>> {
let address = [42; 32];
let pull_request = PullRequest::new(address);
let bytes = pull_request.to_bytes();
let mut socket = tokio::net::TcpStream::connect("127.0.0.1:9000").await?;
// DH temporary: the provider's client port is not in the topology, but we can't change that
// right now without messing up the existing Go mixnet. So I'm going to hardcode this
// for the moment until the Go mixnet goes away.
let provider_socket = SocketAddrV4::new(*provider.ip(), 9000);
println!("Provider: {:?}", provider_socket);
let mut socket = tokio::net::TcpStream::connect(provider_socket).await?;
println!("keep alive: {:?}", socket.keepalive());
socket.set_keepalive(Some(Duration::from_secs(2))).unwrap();
socket.write_all(&bytes[..]).await?;
src/commands/run.rs
+9 -116
View File
@@ -1,123 +1,16 @@
use crate::clients::directory;
use crate::clients::directory::presence::Topology;
use crate::clients::directory::requests::presence_topology_get::PresenceTopologyGetRequester;
use crate::clients::directory::DirectoryClient;
use crate::clients::mix::MixClient;
use crate::clients::provider::ProviderClient;
use crate::banner;
use crate::clients::NymClient;
use crate::utils::bytes;
use base64;
use crate::persistence::pemstore;
use clap::ArgMatches;
use curve25519_dalek::montgomery::MontgomeryPoint;
use sphinx::route::Destination;
use sphinx::route::Node as SphinxNode;
use std::net::SocketAddrV4;
use std::time::Duration;
use tokio::runtime::Runtime;
use tokio::time::{interval_at, Instant};
pub fn execute(matches: &ArgMatches) {
unimplemented!() // currently the 'run' starts websocket!
println!("{}", banner());
// let is_local = matches.is_present("local");
// println!("Starting client, local: {:?}", is_local);
//
// // todo: to be taken from config or something
// let my_address = [42u8; 32];
// let is_local = true;
// let client = NymClient::new(my_address, is_local);
// client.start().unwrap();
// Grab the network topology from the remote directory server
// let topology = get_topology(is_local);
let is_local = matches.is_present("local");
let id = matches.value_of("id").unwrap().to_string();
println!("Starting client...");
// // Grab the network topology from the remote directory server
// let topology = get_topology();
//
// // Create the runtime, probably later move it to Client struct itself?
// let mut rt = Runtime::new().unwrap();
//
// // Spawn the root task
// rt.block_on(async {
// let start = Instant::now() + Duration::from_nanos(1000);
// let mut interval = interval_at(start, Duration::from_millis(1000));
// let mut i: usize = 0;
// loop {
// interval.tick().await;
//// let message = format!("Hello, Sphinx {}", i).as_bytes().to_vec();
////
//// let route_len = 2;
////
//// // data needed to generate a new Sphinx packet
//// let route = route_from(&topology, route_len);
//// let destination = get_destination();
//// let delays = sphinx::header::delays::generate(route_len);
////
//// // build the packet
//// let packet =
//// sphinx::SphinxPacket::new(message, &route[..], &destination, &delays).unwrap();
////
//// // send to mixnet
//// let mix_client = MixClient::new();
//// let result = mix_client.send(packet, route.first().unwrap()).await;
//// println!("packet sent: {:?}", i);
// i += 1;
//
// // retrieve messages every now and then
// if i % 3 == 0 {
// interval.tick().await;
// println!("going to retrieve messages!");
// let provider_client = ProviderClient::new();
// provider_client.retrieve_messages().await.unwrap();
// }
// }
// })
// // Create the runtime, probably later move it to Client struct itself?
// let mut rt = Runtime::new().unwrap();
//
// // Spawn the root task
// rt.block_on(async {
// let start = Instant::now() + Duration::from_nanos(1000);
// let mut interval = interval_at(start, Duration::from_millis(1000));
// let mut i: usize = 0;
// loop {
// interval.tick().await;
// let message = format!("Hello, Sphinx {}", i).as_bytes().to_vec();
//
// let route_len = 2;
//
// // data needed to generate a new Sphinx packet
// let route = route_from(&topology, route_len);
// let destination = get_destination();
// let delays = sphinx::header::delays::generate(route_len);
//
// // build the packet
// let packet =
// sphinx::SphinxPacket::new(message, &route[..], &destination, &delays).unwrap();
//
// // send to mixnet
// let mix_client = MixClient::new();
// mix_client
// .send(packet, route.first().unwrap())
// .await
// .unwrap();
// println!("packet sent: {:?}", i);
// i += 1;
//
// // retrieve messages every now and then
// if i % 3 == 0 {
// interval.tick().await;
// println!("going to retrieve messages!");
// let provider_client = ProviderClient::new();
// provider_client.retrieve_messages().await.unwrap();
// }
// }
// })
}
// TODO: where do we retrieve this guy from?
fn get_destination() -> Destination {
Destination {
address: [42u8; 32],
identifier: [1u8; 16],
}
let keypair = pemstore::read_keypair_from_disk(id);
let client = NymClient::new(keypair.public_bytes(), is_local);
client.start().unwrap();
}
src/commands/tcpsocket.rs
+25 -2
View File
@@ -1,10 +1,33 @@
use crate::banner;
use crate::clients::NymClient;
use crate::persistence::pemstore;
use crate::sockets::tcp;
use clap::ArgMatches;
use std::net::ToSocketAddrs;
pub fn execute(matches: &ArgMatches) {
let port = match matches.value_of("port").unwrap().parse::<u16>() {
println!("{}", banner());
let is_local = matches.is_present("local");
let id = matches.value_of("id").unwrap().to_string();
let port = match matches.value_of("port").unwrap_or("9001").parse::<u16>() {
Ok(n) => n,
Err(err) => panic!("Invalid port value provided - {:?}", err),
};
println!("On the following port: {:?}", port);
println!("Starting TCP socket on port: {:?}", port);
println!("Listening for messages...");
let socket_address = ("127.0.0.1", port)
.to_socket_addrs()
.expect("Failed to combine host and port")
.next()
.expect("Failed to extract the socket address from the iterator");
let keypair = pemstore::read_keypair_from_disk(id);
let _client = NymClient::new(keypair.public_bytes(), is_local);
// Question: should we be passing the client into the TCP socket somehow next?
tcp::start(socket_address);
}
src/commands/websocket.rs
+12 -7
View File
@@ -1,15 +1,21 @@
use crate::banner;
use crate::clients::NymClient;
use crate::persistence::pemstore;
use crate::sockets::ws;
use clap::ArgMatches;
use std::net::ToSocketAddrs;
pub fn execute(matches: &ArgMatches) {
let port = match matches.value_of("port").unwrap().parse::<u16>() {
println!("{}", banner());
let is_local = matches.is_present("local");
let id = matches.value_of("id").unwrap().to_string();
let port = match matches.value_of("port").unwrap_or("9001").parse::<u16>() {
Ok(n) => n,
Err(err) => panic!("Invalid port value provided - {:?}", err),
};
println!("{}", banner());
println!("Starting websocket on port: {:?}", port);
println!("Listening for messages...");
@@ -19,13 +25,12 @@ pub fn execute(matches: &ArgMatches) {
.next()
.expect("Failed to extract the socket address from the iterator");
let is_local = matches.is_present("local");
println!("Starting client, local: {:?}", is_local);
// todo: to be taken from config or something
let my_address = [42u8; 32];
let is_local = true;
let client = NymClient::new(my_address, is_local);
let keypair = pemstore::read_keypair_from_disk(id);
let client = NymClient::new(keypair.public_bytes(), is_local);
client.start(socket_address).unwrap();
}
src/identity/mixnet.rs
+4 -4
View File
@@ -23,11 +23,11 @@ impl KeyPair {
KeyPair { private, public }
}
pub fn private_bytes(&self) -> Vec<u8> {
self.private.to_bytes().to_vec()
pub fn private_bytes(&self) -> [u8; 32] {
self.private.to_bytes()
}
pub fn public_bytes(&self) -> Vec<u8> {
self.public.to_bytes().to_vec()
pub fn public_bytes(&self) -> [u8; 32] {
self.public.to_bytes()
}
}
src/main.rs
+24 -2
View File
@@ -31,6 +31,12 @@ fn main() {
.subcommand(
SubCommand::with_name("run")
.about("Run a persistent Nym client process")
.arg(Arg::with_name("id")
.long("id")
.help("Id of the nym-mixnet-client we want to run.")
.takes_value(true)
.required(true)
)
.arg(Arg::with_name("local")
.long("local")
.help("Flag to indicate whether the client is expected to run on the local deployment.")
@@ -44,9 +50,19 @@ fn main() {
Arg::with_name("port")
.short("p")
.long("port")
.help("Port to listen on")
.help("Port for TCP socket to listen on")
.takes_value(true)
.required(true),
).arg(Arg::with_name("local")
.long("local")
.help("Flag to indicate whether the client is expected to run on the local deployment.")
.takes_value(false)
)
.arg(Arg::with_name("id")
.long("id")
.help("Id of the nym-mixnet-client we want to run.")
.takes_value(true)
.required(true)
)
.arg(Arg::with_name("local")
.long("local")
@@ -61,7 +77,7 @@ fn main() {
Arg::with_name("port")
.short("p")
.long("port")
.help("Port to listen on")
.help("Port for websocket to listen on")
.takes_value(true)
.required(true),
)
@@ -70,6 +86,12 @@ fn main() {
.help("Flag to indicate whether the client is expected to run on the local deployment.")
.takes_value(false)
)
.arg(Arg::with_name("id")
.long("id")
.help("Id of the nym-mixnet-client we want to run.")
.takes_value(true)
.required(true)
)
)
.get_matches();
src/persistence/pemstore.rs
+9 -2
View File
@@ -5,6 +5,13 @@ use std::fs::File;
use std::io::prelude::*;
use std::path::PathBuf;
pub fn read_keypair_from_disk(id: String) -> KeyPair {
let pathfinder = Pathfinder::new(id);
let pem_store = PemStore::new(pathfinder);
let keypair = pem_store.read();
keypair
}
pub struct PemStore {
config_dir: PathBuf,
private_mix_key: PathBuf,
@@ -52,10 +59,10 @@ impl PemStore {
);
}
fn write_pem_file(&self, filepath: PathBuf, data: Vec<u8>, tag: String) {
fn write_pem_file(&self, filepath: PathBuf, data: [u8; 32], tag: String) {
let pem = Pem {
tag,
contents: data,
contents: data.to_vec(),
};
let key = encode(&pem);
src/sockets/mod.rs
+1
View File
@@ -1 +1,2 @@
pub mod tcp;
pub mod ws;
src/sockets/tcp.rs
+5
View File
@@ -0,0 +1,5 @@
use std::net::SocketAddr;
pub fn start(_socket_address: SocketAddr) {
println!("TCP server time!!!");
}
src/utils/sphinx.rs
+8 -2
View File
@@ -1,5 +1,6 @@
use crate::clients::directory::presence::Topology;
use crate::utils::{bytes, topology};
use curve25519_dalek::montgomery::MontgomeryPoint;
use sphinx::route::{Destination, DestinationAddressBytes, Node, NodeAddressBytes, SURBIdentifier};
use sphinx::SphinxPacket;
@@ -21,10 +22,15 @@ pub fn encapsulate_message(
topology: &Topology,
) -> (NodeAddressBytes, SphinxPacket) {
let mixes_route = topology::route_from(&topology, 1);
let first_provider = topology.mix_provider_nodes.first().unwrap();
let decoded_key_bytes =
base64::decode_config(&first_provider.pub_key, base64::URL_SAFE).unwrap();
let key_bytes = bytes::zero_pad_to_32(decoded_key_bytes);
let key = MontgomeryPoint(key_bytes);
let provider = Node::new(
topology::socket_bytes_from_string("127.0.0.1:8081".to_string()),
Default::default(),
topology::socket_bytes_from_string(first_provider.host.clone()),
key,
);
let route = [mixes_route, vec![provider]].concat();
src/utils/topology.rs
-2
View File
@@ -2,8 +2,6 @@ use crate::clients::directory;
use crate::clients::directory::presence::Topology;
use crate::clients::directory::requests::presence_topology_get::PresenceTopologyGetRequester;
use crate::clients::directory::DirectoryClient;
use crate::clients::mix::MixClient;
use crate::clients::provider::ProviderClient;
use crate::utils::bytes;
use curve25519_dalek::montgomery::MontgomeryPoint;
use sphinx::route::Node as SphinxNode;