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

Merge branch 'develop' into features/auth_token

Browse Source
This commit is contained in:
Jedrzej Stuczynski
2019-12-18 11:41:13 +00:00
parent 20fe59b9ba 8dd8873a03
commit a8e46d34df
6 changed files with 1471 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1253 -5
View File
File diff suppressed because it is too large Load Diff
Cargo.toml
+5 -3
View File
@@ -7,15 +7,17 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
base64 = "0.11.0"
clap = "2.33.0"
curve25519-dalek = "1.2.3"
hex = "0.4.0"
futures = "0.3.1"
nym-client = { path = "../nym-client" }
rand = "0.7.2"
sfw-provider-requests = { path = "./sfw-provider-requests" }
sphinx = { path = "../sphinx" }
tokio = { version = "0.2.4", features = ["full"] }
sha2 = "0.8.0"
serde = "1.0"
serde_json = "1.0"
hmac = "0.7.1"
serde = { version = "1.0.104", features = ["derive"] }
serde_json = "1.0.44"
hmac = "0.7.1"
src/main.rs
+52 -34
View File
@@ -1,11 +1,13 @@
use crate::provider::presence;
use crate::provider::ServiceProvider;
use clap::{App, Arg, ArgMatches, SubCommand};
use curve25519_dalek::scalar::Scalar;
use std::net::ToSocketAddrs;
use std::path::PathBuf;
use std::process;
use std::thread;
mod provider;
pub mod provider;
fn main() {
let arg_matches = App::new("Nym Service Provider")
@@ -26,7 +28,6 @@ fn main() {
.long("mixPort")
.help("The port on which the service provider will be listening for sphinx packets")
.takes_value(true)
.required(true),
)
.arg(
Arg::with_name("clientHost")
@@ -39,7 +40,6 @@ fn main() {
.long("clientPort")
.help("The port on which the service provider will be listening for client sfw-provider-requests")
.takes_value(true)
.required(true),
)
.arg(
Arg::with_name("keyfile")
@@ -63,8 +63,11 @@ fn main() {
.help("Directory of the ledger of registered clients")
.takes_value(true)
.required(true),
),
).arg(Arg::with_name("local")
.long("local")
.help("Flag to indicate whether the provider should run on a local deployment.")
.takes_value(false)
)
)
.get_matches();
@@ -75,33 +78,54 @@ fn main() {
}
fn run(matches: &ArgMatches) {
let config = new_config(matches);
let provider = ServiceProvider::new(&config);
// Start sending presence notifications in a separate thread
thread::spawn(move || {
let notifier = presence::Notifier::new(&config);
notifier.run();
});
provider.start_listening().unwrap()
}
fn execute(matches: ArgMatches) -> Result<(), String> {
match matches.subcommand() {
("run", Some(m)) => Ok(run(m)),
_ => Err(usage()),
}
}
fn new_config(matches: &ArgMatches) -> provider::Config {
println!("Running the service provider!");
let is_local = matches.is_present("local");
let directory_server = if is_local {
"http://localhost:8080".to_string()
} else {
"https://directory.nymtech.net".to_string()
};
let mix_host = matches.value_of("mixHost").unwrap_or("0.0.0.0");
let mix_port = match matches.value_of("mixPort").unwrap().parse::<u16>() {
let mix_port = match matches.value_of("mixPort").unwrap_or("8085").parse::<u16>() {
Ok(n) => n,
Err(err) => panic!("Invalid port value provided - {:?}", err),
};
let client_host = matches.value_of("clientHost").unwrap_or("0.0.0.0");
let client_port = match matches.value_of("clientPort").unwrap().parse::<u16>() {
let client_port = match matches
.value_of("clientPort")
.unwrap_or("9000")
.parse::<u16>()
{
Ok(n) => n,
Err(err) => panic!("Invalid port value provided - {:?}", err),
};
let secret_key: Scalar = match matches.value_of("keyfile") {
Some(keyfile) => {
println!("Todo: load keyfile from <{:?}>", keyfile);
Default::default()
}
None => {
println!("Todo: generate fresh sphinx keypair");
Default::default()
}
};
let store_dir = PathBuf::from(matches.value_of("storeDir").unwrap());
let store_dir = PathBuf::from(matches.value_of("storeDir").unwrap_or("/tmp/nym-provider"));
let registered_client_ledger_dir = PathBuf::from(matches.value_of("registeredLedger").unwrap());
let (secret_key, public_key) = sphinx::crypto::keygen();
println!("The value of mix_host is: {:?}", mix_host);
println!("The value of mix_port is: {:?}", mix_port);
@@ -109,7 +133,10 @@ fn run(matches: &ArgMatches) {
println!("The value of client_port is: {:?}", client_port);
println!("The value of key is: {:?}", secret_key);
println!("The value of store_dir is: {:?}", store_dir);
println!("The value of registered_client_ledger_dir is: {:?}", registered_client_ledger_dir);
println!(
"The value of registered_client_ledger_dir is: {:?}",
registered_client_ledger_dir
);
let mix_socket_address = (mix_host, mix_port)
.to_socket_addrs()
@@ -132,22 +159,13 @@ fn run(matches: &ArgMatches) {
client_socket_address
);
// make sure our socket_address is equal to our predefined-hardcoded value
assert_eq!("127.0.0.1:8081", mix_socket_address.to_string());
let provider = ServiceProvider::new(
mix_socket_address,
provider::Config {
directory_server,
public_key,
client_socket_address,
mix_socket_address,
secret_key,
store_dir,
);
provider.start_listening().unwrap()
}
fn execute(matches: ArgMatches) -> Result<(), String> {
match matches.subcommand() {
("run", Some(m)) => Ok(run(m)),
_ => Err(usage()),
store_dir: PathBuf::from(store_dir),
}
}
src/provider/mod.rs
+110 -38
View File
@@ -3,6 +3,7 @@ use std::path::PathBuf;
use std::sync::Arc;
use std::sync::RwLock;
use curve25519_dalek::montgomery::MontgomeryPoint;
use curve25519_dalek::scalar::Scalar;
use tokio::prelude::*;
use tokio::runtime::Runtime;
@@ -10,19 +11,62 @@ use tokio::runtime::Runtime;
use crate::provider::client_handling::{ClientProcessingData, ClientRequestProcessor};
use crate::provider::mix_handling::{MixPacketProcessor, MixProcessingData};
use crate::provider::storage::ClientStorage;
use std::collections::HashMap;
use sphinx::route::DestinationAddressBytes;
use futures::io::Error;
use sfw_provider_requests::requests::AuthToken;
use sphinx::route::DestinationAddressBytes;
use std::collections::HashMap;
mod client_handling;
mod mix_handling;
pub mod presence;
mod storage;
// TODO: if we ever create config file, this should go there
const STORED_MESSAGE_FILENAME_LENGTH: usize = 16;
const MESSAGE_RETRIEVAL_LIMIT: usize = 2;
pub struct Config {
pub client_socket_address: SocketAddr,
pub directory_server: String,
pub mix_socket_address: SocketAddr,
pub public_key: MontgomeryPoint,
pub secret_key: Scalar,
pub store_dir: PathBuf,
}
impl Config {
pub fn public_key_string(&self) -> String {
let key_bytes = self.public_key.to_bytes().to_vec();
base64::encode_config(&key_bytes, base64::URL_SAFE)
}
}
#[derive(Debug)]
pub enum ProviderError {
TcpListenerBindingError,
TcpListenerConnectionError,
TcpListenerUnexpectedEof,
TcpListenerUnknownError,
}
impl From<io::Error> for ProviderError {
fn from(err: Error) -> Self {
use ProviderError::*;
match err.kind() {
io::ErrorKind::ConnectionRefused => TcpListenerConnectionError,
io::ErrorKind::ConnectionReset => TcpListenerConnectionError,
io::ErrorKind::ConnectionAborted => TcpListenerConnectionError,
io::ErrorKind::NotConnected => TcpListenerConnectionError,
io::ErrorKind::AddrInUse => TcpListenerBindingError,
io::ErrorKind::AddrNotAvailable => TcpListenerBindingError,
io::ErrorKind::UnexpectedEof => TcpListenerUnexpectedEof,
_ => TcpListenerUnknownError,
}
}
}
pub struct ServiceProvider {
mix_network_address: SocketAddr,
client_network_address: SocketAddr,
@@ -32,20 +76,21 @@ pub struct ServiceProvider {
}
impl ServiceProvider {
pub fn new(mix_network_address: SocketAddr, client_network_address: SocketAddr, secret_key: Scalar, store_dir: PathBuf) -> Self {
let mut registered_clients_ledger = HashMap::new();
pub fn new(config: &Config) -> Self {
ServiceProvider {
mix_network_address,
client_network_address,
secret_key,
store_dir,
registered_clients_ledger,
mix_network_address: config.mix_socket_address,
client_network_address: config.client_socket_address,
secret_key: config.secret_key,
store_dir: PathBuf::from(config.store_dir.clone()),
// TODO: load initial ledger from file
registered_clients_ledger: HashMap::new(),
}
}
async fn process_mixnet_socket_connection(mut socket: tokio::net::TcpStream, processing_data: Arc<RwLock<MixProcessingData>>) {
async fn process_mixnet_socket_connection(
mut socket: tokio::net::TcpStream,
processing_data: Arc<RwLock<MixProcessingData>>,
) {
let mut buf = [0u8; sphinx::PACKET_SIZE];
// In a loop, read data from the socket and write the data back.
@@ -57,14 +102,21 @@ impl ServiceProvider {
return;
}
Ok(_) => {
let store_data = match MixPacketProcessor::process_sphinx_data_packet(buf.as_ref(), processing_data.as_ref()) {
let store_data = match MixPacketProcessor::process_sphinx_data_packet(
buf.as_ref(),
processing_data.as_ref(),
) {
Ok(sd) => sd,
Err(e) => {
eprintln!("failed to process sphinx packet; err = {:?}", e);
return;
}
};
ClientStorage::store_processed_data(store_data, processing_data.read().unwrap().store_dir.as_path()).unwrap_or_else(|e| {
ClientStorage::store_processed_data(
store_data,
processing_data.read().unwrap().store_dir.as_path(),
)
.unwrap_or_else(|e| {
eprintln!("failed to store processed sphinx message; err = {:?}", e);
return;
});
@@ -93,7 +145,10 @@ impl ServiceProvider {
}
// TODO: FIGURE OUT HOW TO SET READ_DEADLINES IN TOKIO
async fn process_client_socket_connection(mut socket: tokio::net::TcpStream, processing_data: Arc<RwLock<ClientProcessingData>>, secret_key: Scalar) {
async fn process_client_socket_connection(
mut socket: tokio::net::TcpStream,
processing_data: Arc<RwLock<ClientProcessingData>>,
) {
let mut buf = [0; 1024];
// TODO: restore the for loop once we go back to persistent tcp socket connection
@@ -104,7 +159,10 @@ impl ServiceProvider {
Err(())
}
Ok(n) => {
match ClientRequestProcessor::process_client_request(buf[..n].as_ref(), processing_data.as_ref(), secret_key) {
match ClientRequestProcessor::process_client_request(
buf[..n].as_ref(),
processing_data.as_ref(),
) {
Err(e) => {
eprintln!("failed to process client request; err = {:?}", e);
Err(())
@@ -134,9 +192,13 @@ impl ServiceProvider {
}
}
async fn start_mixnet_listening(&self) -> Result<(), Box<dyn std::error::Error>> {
let mut listener = tokio::net::TcpListener::bind(self.mix_network_address).await?;
let processing_data = MixProcessingData::new(self.secret_key, self.store_dir.clone()).add_arc_rwlock();
async fn start_mixnet_listening(
address: SocketAddr,
secret_key: Scalar,
store_dir: PathBuf,
) -> Result<(), ProviderError> {
let mut listener = tokio::net::TcpListener::bind(address).await?;
let processing_data = MixProcessingData::new(secret_key, store_dir).add_arc_rwlock();
loop {
let (socket, _) = listener.accept().await?;
@@ -144,41 +206,55 @@ impl ServiceProvider {
// (if I understand it all correctly)
let thread_processing_data = processing_data.clone();
tokio::spawn(async move {
ServiceProvider::process_mixnet_socket_connection(socket, thread_processing_data).await
ServiceProvider::process_mixnet_socket_connection(socket, thread_processing_data)
.await
});
}
}
async fn start_client_listening(&self) -> Result<(), Box<dyn std::error::Error>> {
let mut listener = tokio::net::TcpListener::bind(self.client_network_address).await?;
let processing_data = ClientProcessingData::new(self.store_dir.clone(), self.registered_clients_ledger.clone()).add_arc_rwlock();
let key = self.secret_key.clone();
async fn start_client_listening(
address: SocketAddr,
store_dir: PathBuf,
client_ledger: HashMap<AuthToken, DestinationAddressBytes>,
secret_key: Scalar,
) -> Result<(), ProviderError> {
let mut listener = tokio::net::TcpListener::bind(address).await?;
let processing_data =
ClientProcessingData::new(store_dir, client_ledger, secret_key).add_arc_rwlock();
loop {
let (socket, _) = listener.accept().await?;
// do note that the underlying data is NOT copied here; arc is incremented and lock is shared
// (if I understand it all correctly)
let thread_processing_data = processing_data.clone();
tokio::spawn(async move {
ServiceProvider::process_client_socket_connection(socket, thread_processing_data, key).await
ServiceProvider::process_client_socket_connection(socket, thread_processing_data)
.await
});
}
}
async fn start_listeners(&self) -> (Result<(), Box<dyn std::error::Error>>, Result<(), Box<dyn std::error::Error>>) {
futures::future::join(self.start_mixnet_listening(), self.start_client_listening()).await
}
pub fn start_listening(&self) -> Result<(), Box<dyn std::error::Error>> {
// Create the runtime, probably later move it to Provider struct itself?
// TODO: figure out the difference between Runtime and Handle
let mut rt = Runtime::new()?;
// let mut h = rt.handle();
// let mut h = rt.handle();
let mix_future = rt.spawn(ServiceProvider::start_mixnet_listening(
self.mix_network_address,
self.secret_key,
self.store_dir.clone(),
));
let client_future = rt.spawn(ServiceProvider::start_client_listening(
self.client_network_address,
self.store_dir.clone(),
self.registered_clients_ledger.clone(), // we're just cloning the initial ledger state
self.secret_key,
));
// Spawn the root task
rt.block_on(async {
let future_results = self.start_listeners().await;
assert!(future_results.0.is_ok() && future_results.1.is_ok())
let future_results = futures::future::join(mix_future, client_future).await;
assert!(future_results.0.is_ok() && future_results.1.is_ok());
});
// this line in theory should never be reached as the runtime should be permanently blocked on listeners
@@ -186,7 +262,3 @@ impl ServiceProvider {
Ok(())
}
}
src/provider/presence.rs
+46
View File
@@ -0,0 +1,46 @@
use crate::provider;
use nym_client::clients::directory;
use nym_client::clients::directory::DirectoryClient;
use std::thread;
use std::time::Duration;
use nym_client::clients::directory::presence::MixProviderPresence;
use nym_client::clients::directory::requests::presence_providers_post::PresenceMixProviderPoster;
pub struct Notifier {
pub net_client: directory::Client,
presence: MixProviderPresence,
}
impl Notifier {
pub fn new(config: &provider::Config) -> Notifier {
let directory_config = directory::Config {
base_url: config.directory_server.clone(),
};
let net_client = directory::Client::new(directory_config);
let presence = MixProviderPresence {
host: config.mix_socket_address.to_string(),
pub_key: config.public_key_string(),
registered_clients: vec![],
};
Notifier {
net_client,
presence,
}
}
pub fn notify(&self) {
self.net_client
.presence_providers_post
.post(&self.presence)
.unwrap();
}
pub fn run(&self) {
loop {
self.notify();
thread::sleep(Duration::from_secs(5));
}
}
}
src/provider/storage/mod.rs
+5
View File
@@ -66,6 +66,11 @@ impl ClientStorage {
store_data.message, full_store_path
);
println!(
"string message: {:?}",
std::str::from_utf8(&store_data.message)
);
// TODO: what to do with surbIDs??
// we can use normal io here, no need for tokio as it's all happening in one thread per connection