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

Merge branch 'develop' into feature/provider-from-topology

Browse Source
This commit is contained in:
Dave Hrycyszyn
2019-12-17 13:38:02 +00:00
parent c508d1518d 273e7df39a
commit 253dbdafaf
12 changed files with 440 additions and 113 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(Debug, Deserialize, Serialize)]
#[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(Debug, Deserialize, Serialize)]
#[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(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[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(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[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(Debug, Deserialize, Serialize)]
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Topology {
pub coco_nodes: Vec<CocoPresence>,
src/clients/mix.rs
+5 -3
View File
@@ -1,5 +1,6 @@
use sphinx::route::Node as MixNode;
use sphinx::route::{Node as MixNode, NodeAddressBytes};
use sphinx::SphinxPacket;
use std::net::SocketAddr;
use std::net::{Ipv4Addr, SocketAddrV4};
use tokio::prelude::*;
@@ -14,14 +15,15 @@ impl MixClient {
pub async fn send(
&self,
packet: SphinxPacket,
mix: &MixNode,
mix_addr: NodeAddressBytes,
) -> Result<(), Box<dyn std::error::Error>> {
let bytes = packet.to_bytes();
let b = mix.address;
let b = mix_addr;
let host = Ipv4Addr::new(b[0], b[1], b[2], b[3]);
let port: u16 = u16::from_be_bytes([b[4], b[5]]);
let socket_address = SocketAddrV4::new(host, port);
println!("socket addr: {:?}", socket_address);
let mut stream = tokio::net::TcpStream::connect(socket_address).await?;
stream.write_all(&bytes[..]).await?;
src/clients/mod.rs
+183
View File
@@ -1,4 +1,187 @@
use crate::clients::directory::presence::Topology;
use crate::clients::mix::MixClient;
use crate::clients::provider::ProviderClient;
use crate::utils;
use crate::utils::topology::get_topology;
use futures::channel::mpsc;
use futures::select;
use futures::{SinkExt, StreamExt};
use sphinx::route::{Destination, DestinationAddressBytes, NodeAddressBytes};
use sphinx::SphinxPacket;
use std::net::SocketAddrV4;
use std::time::Duration;
use tokio::runtime::Runtime;
pub mod directory;
pub mod mix;
pub mod provider;
pub mod validator;
// TODO: put that in config once it exists
const LOOP_COVER_AVERAGE_DELAY: f64 = 1.0;
// assume seconds
const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 10.0;
// assume seconds;
const FETCH_MESSAGES_DELAY: f64 = 10.0; // assume seconds;
// provider-poller sends polls service provider; receives messages
// provider-poller sends (TX) to ReceivedBufferController (RX)
// ReceivedBufferController sends (TX) to ... ??Client??
// outQueueController sends (TX) to TrafficStreamController (RX)
// TrafficStreamController sends messages to mixnet
// ... ??Client?? sends (TX) to outQueueController (RX)
// Loop cover traffic stream just sends messages to mixnet without any channel communication
struct MixMessage(NodeAddressBytes, SphinxPacket);
struct MixTrafficController;
impl MixTrafficController {
// this was way more difficult to implement than what this code may suggest...
async fn run(mut rx: mpsc::UnboundedReceiver<MixMessage>) {
let mix_client = MixClient::new();
while let Some(mix_message) = rx.next().await {
println!(
"[MIX TRAFFIC CONTROL] - got a mix_message for {:?}",
mix_message.0
);
let send_res = mix_client.send(mix_message.1, mix_message.0).await;
match send_res {
Ok(_) => println!("We successfully sent the message!"),
Err(e) => eprintln!("We failed to send the message :( - {:?}", e),
};
}
}
}
pub struct NymClient {
// to be replaced by something else I guess
address: DestinationAddressBytes,
pub input_tx: mpsc::UnboundedSender<InputMessage>,
// to be used by "send" function or socket, etc
input_rx: mpsc::UnboundedReceiver<InputMessage>,
is_local: bool,
}
#[derive(Debug)]
pub struct InputMessage(pub Destination, pub Vec<u8>);
impl NymClient {
pub fn new(address: DestinationAddressBytes, is_local: bool) -> Self {
let (input_tx, input_rx) = mpsc::unbounded::<InputMessage>();
NymClient {
address,
input_tx,
input_rx,
is_local,
}
}
async fn start_loop_cover_traffic_stream(
mut tx: mpsc::UnboundedSender<MixMessage>,
our_info: Destination,
topology: &Topology,
) -> Result<(), Box<dyn std::error::Error>> {
loop {
println!("[LOOP COVER TRAFFIC STREAM] - next cover message!");
let delay = utils::poisson::sample(LOOP_COVER_AVERAGE_DELAY);
let delay_duration = Duration::from_secs_f64(delay);
tokio::time::delay_for(delay_duration).await;
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?;
}
}
async fn control_out_queue(
mut mix_tx: mpsc::UnboundedSender<MixMessage>,
mut input_rx: mpsc::UnboundedReceiver<InputMessage>,
our_info: Destination,
topology: &Topology,
) -> Result<(), Box<dyn std::error::Error>> {
loop {
println!("[OUT QUEUE] here I will be sending real traffic (or loop cover if nothing is available)");
select! {
real_message = input_rx.next() => {
println!("[OUT QUEUE] - we got a real message!");
let real_message = real_message.expect("The channel must have closed! - if the client hasn't crashed, it should have!");
println!("real: {:?}", real_message);
let encapsulated_message = utils::sphinx::encapsulate_message(real_message.0, real_message.1, topology);
mix_tx.send(MixMessage(encapsulated_message.0, encapsulated_message.1)).await?;
},
default => {
println!("[OUT QUEUE] - no real message - going to send extra loop cover");
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?;
}
};
let delay_duration = Duration::from_secs_f64(MESSAGE_SENDING_AVERAGE_DELAY);
tokio::time::delay_for(delay_duration).await;
}
}
async fn start_provider_polling(
provider_address: SocketAddrV4,
) -> Result<(), Box<dyn std::error::Error>> {
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();
}
}
pub fn start(self) -> Result<(), Box<dyn std::error::Error>> {
println!("starting nym client");
let (mix_tx, mix_rx) = mpsc::unbounded();
let mut rt = Runtime::new()?;
rt.spawn(MixTrafficController::run(mix_rx));
let topology = get_topology(self.is_local);
let provider_address: SocketAddrV4 = topology
.mix_provider_nodes
.first()
.unwrap()
.host
.parse()
.unwrap();
rt.block_on(async {
let future_results = futures::future::join3(
NymClient::start_loop_cover_traffic_stream(
mix_tx.clone(),
Destination::new(self.address, Default::default()),
&topology.clone(),
),
NymClient::control_out_queue(
mix_tx,
self.input_rx,
Destination::new(self.address, Default::default()),
&topology.clone(),
),
NymClient::start_provider_polling(provider_address),
)
.await;
// start websocket handler here
assert!(
future_results.0.is_ok() && future_results.1.is_ok() && future_results.2.is_ok()
);
});
// this line in theory should never be reached as the runtime should be permanently blocked on traffic senders
eprintln!("The client went kaput...");
Ok(())
}
}
src/clients/provider.rs
+8 -2
View File
@@ -14,13 +14,19 @@ impl ProviderClient {
pub async fn retrieve_messages(
&self,
provider: &SocketAddrV4,
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(provider).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
+6 -103
View File
@@ -4,6 +4,7 @@ use crate::clients::directory::requests::presence_topology_get::PresenceTopology
use crate::clients::directory::DirectoryClient;
use crate::clients::mix::MixClient;
use crate::clients::provider::ProviderClient;
use crate::clients::NymClient;
use crate::utils::bytes;
use base64;
use clap::ArgMatches;
@@ -19,110 +20,12 @@ pub fn execute(matches: &ArgMatches) {
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 provider_address: SocketAddrV4 = topology
.mix_provider_nodes
.first()
.unwrap()
.host
.parse()
.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 = 1; // TODO: kill magic number
// 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(&provider_address)
.await
.unwrap();
}
}
})
}
fn get_topology(is_local: bool) -> Topology {
let url = if is_local {
"http://localhost:8080".to_string()
} else {
"https://directory.nymtech.net".to_string()
};
println!("Using directory server: {:?}", url);
let directory_config = directory::Config { base_url: url };
let directory = directory::Client::new(directory_config);
let topology = directory
.presence_topology
.get()
.expect("Failed to retrieve network topology.");
topology
}
fn route_from(topology: &Topology, route_len: usize) -> Vec<SphinxNode> {
let mut route = vec![];
let nodes = topology.mix_nodes.iter();
for mix in nodes.take(route_len) {
let address_bytes = socket_bytes_from_string(mix.host.clone());
let decoded_key_bytes = base64::decode_config(&mix.pub_key, base64::URL_SAFE).unwrap();
let key_bytes = bytes::zero_pad_to_32(decoded_key_bytes);
let key = MontgomeryPoint(key_bytes);
let sphinx_node = SphinxNode {
address: address_bytes,
pub_key: key,
};
route.push(sphinx_node);
}
route
}
// Parses a String socket address, then returns it as a
// 4+2 byte array, zero-padded to a constant size of 32 bytes.
fn socket_bytes_from_string(address: String) -> [u8; 32] {
let socket: SocketAddrV4 = address.parse().unwrap();
let host_bytes = socket.ip().octets();
let port_bytes = socket.port().to_be_bytes();
let mut address_bytes = [0u8; 32];
address_bytes[0] = host_bytes[0];
address_bytes[1] = host_bytes[1];
address_bytes[2] = host_bytes[2];
address_bytes[3] = host_bytes[3];
address_bytes[4] = port_bytes[0];
address_bytes[5] = port_bytes[1];
address_bytes
}
// TODO: where do we retrieve this guy from?
src/lib.rs
+1
View File
@@ -1,3 +1,4 @@
pub mod clients;
pub mod identity;
pub mod persistence;
pub mod utils;
src/utils/mod.rs
+3
View File
@@ -1 +1,4 @@
pub mod bytes;
pub mod poisson;
pub mod sphinx;
pub mod topology;
src/utils/poisson.rs
+6
View File
@@ -0,0 +1,6 @@
use rand_distr::{Distribution, Exp};
pub fn sample(average_delay: f64) -> f64 {
let exp = Exp::new(1.0 / average_delay).unwrap();
exp.sample(&mut rand::thread_rng())
}
src/utils/sphinx.rs
+48
View File
@@ -0,0 +1,48 @@
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;
const LOOP_COVER_MESSAGE_PAYLOAD: &[u8] = b"The cake is a lie!";
pub fn loop_cover_message(
our_address: DestinationAddressBytes,
surb_id: SURBIdentifier,
topology: &Topology,
) -> (NodeAddressBytes, SphinxPacket) {
let destination = Destination::new(our_address, surb_id);
encapsulate_message(destination, LOOP_COVER_MESSAGE_PAYLOAD.to_vec(), topology)
}
pub fn encapsulate_message(
recipient: Destination,
message: Vec<u8>,
topology: &Topology,
) -> (NodeAddressBytes, SphinxPacket) {
// here we would be getting topology, etc
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(first_provider.host.clone()),
key,
);
let route = [mixes_route, vec![provider]].concat();
let delays = sphinx::header::delays::generate(route.len());
// build the packet
let packet = sphinx::SphinxPacket::new(message, &route[..], &recipient, &delays).unwrap();
let first_node_address = route.first().unwrap().address;
(first_node_address, packet)
}
src/utils/topology.rs
+61
View File
@@ -0,0 +1,61 @@
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;
use std::net::SocketAddrV4;
use std::string::ToString;
pub(crate) fn get_topology(is_local: bool) -> Topology {
let url = if is_local {
"http://localhost:8080".to_string()
} else {
"https://directory.nymtech.net".to_string()
};
println!("Using directory server: {:?}", url);
let directory_config = directory::Config { base_url: url };
let directory = directory::Client::new(directory_config);
let topology = directory
.presence_topology
.get()
.expect("Failed to retrieve network topology.");
topology
}
pub(crate) fn route_from(topology: &Topology, route_len: usize) -> Vec<SphinxNode> {
let mut route = vec![];
let nodes = topology.mix_nodes.iter();
for mix in nodes.take(route_len) {
let address_bytes = socket_bytes_from_string(mix.host.clone());
let decoded_key_bytes = base64::decode_config(&mix.pub_key, base64::URL_SAFE).unwrap();
let key_bytes = bytes::zero_pad_to_32(decoded_key_bytes);
let key = MontgomeryPoint(key_bytes);
let sphinx_node = SphinxNode {
address: address_bytes,
pub_key: key,
};
route.push(sphinx_node);
}
route
}
pub(crate) fn socket_bytes_from_string(address: String) -> [u8; 32] {
let socket: SocketAddrV4 = address.parse().unwrap();
let host_bytes = socket.ip().octets();
let port_bytes = socket.port().to_be_bytes();
let mut address_bytes = [0u8; 32];
address_bytes[0] = host_bytes[0];
address_bytes[1] = host_bytes[1];
address_bytes[2] = host_bytes[2];
address_bytes[3] = host_bytes[3];
address_bytes[4] = port_bytes[0];
address_bytes[5] = port_bytes[1];
address_bytes
}