From 836cff226a147b53614b45b9ed8a48563d5c3d41 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Mon, 9 Dec 2019 12:47:38 +0000 Subject: [PATCH 1/9] Moved listening loop to provider struct impl --- src/main.rs | 6 ++-- src/provider/mod.rs | 88 +++++++++++++++++++++++++++------------------ 2 files changed, 57 insertions(+), 37 deletions(-) diff --git a/src/main.rs b/src/main.rs index 07204e08d5..3dfc3db735 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,3 +1,4 @@ +use crate::provider::ServiceProvider; use clap::{App, Arg, ArgMatches, SubCommand}; use curve25519_dalek::scalar::Scalar; use std::net::ToSocketAddrs; @@ -48,7 +49,8 @@ fn run(matches: &ArgMatches) { // make sure our socket_address is equal to our predefined-hardcoded value assert_eq!("127.0.0.1:8081", socket_address.to_string()); - provider::listening_loop().unwrap(); + let provider = ServiceProvider::new(socket_address, secret_key); + provider.start_listening().unwrap() } fn main() { @@ -58,7 +60,7 @@ fn main() { .about("Implementation of the Loopix-based Service Provider") .subcommand( SubCommand::with_name("run") - .about("Starts the mixnode") + .about("Starts the service provider") .arg( Arg::with_name("host") .short("h") diff --git a/src/provider/mod.rs b/src/provider/mod.rs index 3be833af96..914767c3c1 100644 --- a/src/provider/mod.rs +++ b/src/provider/mod.rs @@ -1,47 +1,65 @@ use sphinx::{ProcessedPacket, SphinxPacket}; use tokio::prelude::*; use tokio::runtime::Runtime; +use curve25519_dalek::scalar::Scalar; +use std::net::SocketAddr; -pub fn listening_loop() -> Result<(), Box> { - // Create the runtime, probably later move it to Provider struct itself? - let mut rt = Runtime::new()?; +pub struct ServiceProvider { + network_address: SocketAddr, + secret_key: Scalar, +} - // Spawn the root task - rt.block_on(async { - let my_address = "127.0.0.1:8081"; - let mut listener = tokio::net::TcpListener::bind(my_address).await?; +impl ServiceProvider { + pub fn new(network_address: SocketAddr, secret_key: Scalar) -> Self { + ServiceProvider { + network_address, + secret_key, + } + } - println!("Starting Nym store-and-forward Provider on address {:?}", my_address); - println!("Waiting for input..."); + pub fn start_listening(&self) -> Result<(), Box> { + // Create the runtime, probably later move it to Provider struct itself? + let mut rt = Runtime::new()?; - loop { - let (mut inbound, _) = listener.accept().await?; + // Spawn the root task + rt.block_on(async { + let my_address = "127.0.0.1:8081"; + let mut listener = tokio::net::TcpListener::bind(my_address).await?; - tokio::spawn(async move { - let mut buf = [0; sphinx::PACKET_SIZE]; + println!("Starting Nym store-and-forward Provider on address {:?}", my_address); + println!("Waiting for input..."); - loop { - match inbound.read(&mut buf).await { - Ok(length) if length == 0 => - { - println!("Remote connection closed."); + loop { + let (mut inbound, _) = listener.accept().await?; + + tokio::spawn(async move { + let mut buf = [0; sphinx::PACKET_SIZE]; + + loop { + match inbound.read(&mut buf).await { + Ok(length) if length == 0 => + { + println!("Remote connection closed."); + return; + } + Ok(_) => { + let packet = SphinxPacket::from_bytes(buf.to_vec()).unwrap(); + let payload = match packet.process(Default::default()) { + ProcessedPacket::ProcessedPacketFinalHop(_, _, payload) => Some(payload), + _ => None, + }.unwrap(); + let message = payload.get_content(); + } + Err(e) => { + println!("failed to read from socket; err = {:?}", e); return; } - Ok(_) => { - let packet = SphinxPacket::from_bytes(buf.to_vec()).unwrap(); - let payload = match packet.process(Default::default()) { - ProcessedPacket::ProcessedPacketFinalHop(_, _, payload) => Some(payload), - _ => None, - }.unwrap(); - let message = payload.get_content(); - } - Err(e) => { - println!("failed to read from socket; err = {:?}", e); - return; - } - }; - } - }); - } - }) + }; + } + }); + } + }) + } + } + From 385e41a920f5c2881f7e41205b615465aa3a5964 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Mon, 9 Dec 2019 12:57:18 +0000 Subject: [PATCH 2/9] Added store directory as additional provider argument --- src/main.rs | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/src/main.rs b/src/main.rs index 3dfc3db735..e89641e0b2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,6 +2,7 @@ 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; mod provider; @@ -34,9 +35,12 @@ fn run(matches: &ArgMatches) { } }; + let store_dir = PathBuf::from(matches.value_of("storeDir").unwrap()); + println!("The value of host is: {:?}", host); println!("The value of port is: {:?}", port); println!("The value of key is: {:?}", secret_key); + println!("The value of storeDir is: {:?}", store_dir); let socket_address = (host, port) .to_socket_addrs() @@ -49,7 +53,7 @@ fn run(matches: &ArgMatches) { // make sure our socket_address is equal to our predefined-hardcoded value assert_eq!("127.0.0.1:8081", socket_address.to_string()); - let provider = ServiceProvider::new(socket_address, secret_key); + let provider = ServiceProvider::new(socket_address, secret_key, store_dir); provider.start_listening().unwrap() } @@ -82,6 +86,14 @@ fn main() { .long("keyfile") .help("Optional path to the persistent keyfile of the node") .takes_value(true), + ) + .arg( + Arg::with_name("storeDir") + .short("sdir") + .long("storeDir") + .help("Directory storing all packets for the clients") + .takes_value(true) + .required(true), ), ) .get_matches(); From 2c986efcb1c67b9bb0d7974b7d1264a651be42d2 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 10 Dec 2019 13:05:44 +0000 Subject: [PATCH 3/9] Finally won first fight with the borrow checker for concurrent code --- Cargo.lock | 121 ++++++++++++++++++++++++-- Cargo.toml | 3 +- src/provider/mod.rs | 207 ++++++++++++++++++++++++++++++++++++-------- 3 files changed, 290 insertions(+), 41 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e5653f207c..510d013640 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -240,11 +240,88 @@ name = "fuchsia-zircon-sys" version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" +[[package]] +name = "futures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures-channel" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", +] + [[package]] name = "futures-core" version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" +[[package]] +name = "futures-executor" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures-io" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "futures-macro" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)", + "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures-sink" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "futures-task" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "futures-util" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "memchr 2.2.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)", + "proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)", + "proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)", + "slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)", +] + [[package]] name = "generic-array" version = "0.12.3" @@ -429,8 +506,9 @@ version = "0.1.0" dependencies = [ "clap 2.33.0 (registry+https://github.com/rust-lang/crates.io-index)", "curve25519-dalek 1.2.3 (registry+https://github.com/rust-lang/crates.io-index)", + "futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", "sphinx 0.1.0", - "tokio 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)", + "tokio 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -443,11 +521,31 @@ name = "pin-project-lite" version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" +[[package]] +name = "pin-utils" +version = "0.1.0-alpha.4" +source = "registry+https://github.com/rust-lang/crates.io-index" + [[package]] name = "ppv-lite86" version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" +[[package]] +name = "proc-macro-hack" +version = "0.5.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "proc-macro-nested" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" + [[package]] name = "proc-macro2" version = "1.0.6" @@ -628,7 +726,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" [[package]] name = "syn" -version = "1.0.9" +version = "1.0.11" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", @@ -646,7 +744,7 @@ dependencies = [ [[package]] name = "tokio" -version = "0.2.1" +version = "0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "bytes 0.5.2 (registry+https://github.com/rust-lang/crates.io-index)", @@ -672,7 +770,7 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", - "syn 1.0.9 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -770,7 +868,15 @@ dependencies = [ "checksum fuchsia-cprng 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "a06f77d526c1a601b7c4cdd98f54b5eaabffc14d5f2f0296febdc7f357c6d3ba" "checksum fuchsia-zircon 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "2e9763c69ebaae630ba35f74888db465e49e259ba1bc0eda7d06f4a067615d82" "checksum fuchsia-zircon-sys 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7" +"checksum futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b6f16056ecbb57525ff698bb955162d0cd03bee84e6241c27ff75c08d8ca5987" +"checksum futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "fcae98ca17d102fd8a3603727b9259fcf7fa4239b603d2142926189bc8999b86" "checksum futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "79564c427afefab1dfb3298535b21eda083ef7935b4f0ecbfcb121f0aec10866" +"checksum futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "1e274736563f686a837a0568b478bdabfeaec2dca794b5649b04e2fe1627c231" +"checksum futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "e676577d229e70952ab25f3945795ba5b16d63ca794ca9d2c860e5595d20b5ff" +"checksum futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "52e7c56c15537adb4f76d0b7a76ad131cb4d2f4f32d3b0bcabcbe1c7c5e87764" +"checksum futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "171be33efae63c2d59e6dbba34186fe0d6394fb378069a76dfd80fdcffd43c16" +"checksum futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "0bae52d6b29cf440e298856fec3965ee6fa71b06aa7495178615953fd669e5f9" +"checksum futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "c0d66274fb76985d3c62c886d1da7ac4c0903a8c9f754e8fe0f35a6a6cc39e76" "checksum generic-array 0.12.3 (registry+https://github.com/rust-lang/crates.io-index)" = "c68f0274ae0e023facc3c97b2e00f076be70e254bc851d972503b328db79b2ec" "checksum getrandom 0.1.13 (registry+https://github.com/rust-lang/crates.io-index)" = "e7db7ca94ed4cd01190ceee0d8a8052f08a247aa1b469a7f68c6a3b71afcf407" "checksum hermit-abi 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "307c3c9f937f38e3534b1d6447ecf090cafcc9744e4a6360e8b037b2cf5af120" @@ -793,7 +899,10 @@ dependencies = [ "checksum num_cpus 1.11.1 (registry+https://github.com/rust-lang/crates.io-index)" = "76dac5ed2a876980778b8b85f75a71b6cbf0db0b1232ee12f826bccb00d09d72" "checksum opaque-debug 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)" = "2839e79665f131bdb5782e51f2c6c9599c133c6098982a54c794358bf432529c" "checksum pin-project-lite 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "f0af6cbca0e6e3ce8692ee19fb8d734b641899e07b68eb73e9bbbd32f1703991" +"checksum pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)" = "5894c618ce612a3fa23881b152b608bafb8c56cfc22f434a3ba3120b40f7b587" "checksum ppv-lite86 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)" = "74490b50b9fbe561ac330df47c08f3f33073d2d00c150f719147d7c54522fa1b" +"checksum proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)" = "ecd45702f76d6d3c75a80564378ae228a85f0b59d2f3ed43c91b4a69eb2ebfc5" +"checksum proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "369a6ed065f249a159e06c45752c780bda2fb53c995718f9e484d08daa9eb42e" "checksum proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)" = "9c9e470a8dc4aeae2dee2f335e8f533e2d4b347e1434e5671afc49b054592f27" "checksum quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "053a8c8bcc71fcce321828dc897a98ab9760bef03a4fc36693c231e5b3216cfe" "checksum rand 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)" = "3ae1b169243eaf61759b8475a998f0a385e42042370f3a7dbaf35246eacc8412" @@ -814,9 +923,9 @@ dependencies = [ "checksum strsim 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)" = "8ea5119cdb4c55b55d432abb513a0429384878c15dde60cc77b1c99de1a95a6a" "checksum subtle 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "2d67a5a62ba6e01cb2192ff309324cb4875d0c451d55fe2319433abe7a05a8ee" "checksum subtle 2.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7c65d530b10ccaeac294f349038a597e435b18fb456aadd0840a623f83b9e941" -"checksum syn 1.0.9 (registry+https://github.com/rust-lang/crates.io-index)" = "f89693ae015201f8de93fd96bde2d065f8bfc3f97ce006d5bc9f900b97c0c7c0" +"checksum syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)" = "dff0acdb207ae2fe6d5976617f887eb1e35a2ba52c13c7234c790960cdad9238" "checksum textwrap 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d326610f408c7a4eb6f51c37c330e496b08506c9457c9d34287ecc38809fb060" -"checksum tokio 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "beeef686ef92a222de07e089f455d9f8478bbba9651718f9e4b276babe829082" +"checksum tokio 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)" = "bcced6bb623d4bff3739c176c415f13c418f426395c169c9c3cd9a492c715b16" "checksum tokio-macros 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d5795a71419535c6dcecc9b6ca95bdd3c2d6142f7e8343d7beb9923f129aa87e" "checksum typenum 1.11.2 (registry+https://github.com/rust-lang/crates.io-index)" = "6d2783fe2d6b8c1101136184eb41be8b1ad379e4657050b8aaff0c79ee7575f9" "checksum unicode-width 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)" = "caaa9d531767d1ff2150b9332433f32a24622147e5ebb1f26409d5da67afd479" diff --git a/Cargo.toml b/Cargo.toml index ccc83dffa5..06dc593285 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ edition = "2018" [dependencies] sphinx = { path = "../sphinx" } -tokio = { version = "0.2", features = ["full"] } +tokio = { version = "0.2.4", features = ["full"] } +futures = "0.3.1" curve25519-dalek = "1.2.3" clap = "2.33.0" \ No newline at end of file diff --git a/src/provider/mod.rs b/src/provider/mod.rs index 914767c3c1..5e37fd91bb 100644 --- a/src/provider/mod.rs +++ b/src/provider/mod.rs @@ -1,65 +1,204 @@ +use std::borrow::Borrow; +use std::cell::RefCell; +use std::net::SocketAddr; +use std::path::{PathBuf, Path}; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::RwLock; + +use curve25519_dalek::scalar::Scalar; +use futures::StreamExt; use sphinx::{ProcessedPacket, SphinxPacket}; +use sphinx::route::{DestinationAddressBytes, SURBIdentifier}; use tokio::prelude::*; use tokio::runtime::Runtime; -use curve25519_dalek::scalar::Scalar; -use std::net::SocketAddr; + +// TODO: this will probably need to be moved elsewhere I imagine +// DUPLICATE WITH MIXNODE CODE!!! +#[derive(Debug)] +pub enum MixProcessingError { + SphinxRecoveryError, + ReceivedForwardHopError, + InvalidPayload, + NonMatchingRecipient, + StoreFailure, +} + +impl From for MixProcessingError { + // for time being just have a single error instance for all possible results of sphinx::ProcessingError + fn from(_: sphinx::ProcessingError) -> Self { + use MixProcessingError::*; + + SphinxRecoveryError + } +} + +// ProcessingData defines all data required to correctly unwrap sphinx packets +// Do note that we're copying this struct around and hence the secret_key. +// It might, or might not be, what we want +#[derive(Debug, Clone)] +struct ProcessingData { + secret_key: Scalar, +store_dir: PathBuf, +} + +impl ProcessingData { + fn new(secret_key: Scalar, store_dir: PathBuf) -> Self { + ProcessingData { + secret_key, +store_dir, + } + } + + fn add_arc_rwlock(self) -> Arc> { + Arc::new(RwLock::new(self)) + } +} + +struct StoreData { + client_address: DestinationAddressBytes, + client_surb_id: SURBIdentifier, + message: Vec, +} + + +struct PacketProcessor(()); + +impl PacketProcessor { + fn process_sphinx_data_packet(packet_data: &[u8], processing_data: &ProcessingData) -> Result { + let packet = SphinxPacket::from_bytes(packet_data.to_vec())?; + let (client_address, client_surb_id, payload) = match packet.process(processing_data.secret_key) { + ProcessedPacket::ProcessedPacketFinalHop(client_address, surb_id, payload) => (client_address, surb_id, payload), + _ => return Err(MixProcessingError::ReceivedForwardHopError), + }; + + let (payload_destination, message) = payload.try_recover_destination_and_plaintext().ok_or_else(|| MixProcessingError::InvalidPayload)?; + if client_address != payload_destination { + return Err(MixProcessingError::NonMatchingRecipient); + } + + Ok(StoreData { + client_address, + client_surb_id, + message, + }) + } + + fn store_processed_data(store_data: StoreData, store_dir: &PathBuf) -> Result<(), MixProcessingError> { + Ok(()) + } +} + pub struct ServiceProvider { network_address: SocketAddr, secret_key: Scalar, + store_dir: PathBuf, } impl ServiceProvider { - pub fn new(network_address: SocketAddr, secret_key: Scalar) -> Self { + pub fn new(network_address: SocketAddr, secret_key: Scalar, store_dir: PathBuf) -> Self { ServiceProvider { network_address, secret_key, + store_dir, } } + + async fn process_socket_connection(mut socket: tokio::net::TcpStream, processing_data: Arc>) { + println!("look, we can read shared data here! {:?}", processing_data.read().unwrap()); + // NOTE: processing_data is copied here!! +// let mut buf = [0u8; sphinx::PACKET_SIZE]; +// +// // In a loop, read data from the socket and write the data back. +// loop { +// match socket.read(&mut buf).await { +// // socket closed +// Ok(n) if n == 0 => { +// println!("Remote connection closed."); +// return; +// } +// Ok(_) => { +//// let store_data = PacketProcessor::process_sphinx_data_packet(buf.as_ref(), +//// processing_data.read().unwrap().borrow()).unwrap(); +// } +// Err(e) => { +// println!("failed to read from socket; err = {:?}", e); +// return; +// } +// }; +// +// // Write the some data back +// if let Err(e) = socket.write_all(b"foomp").await { +// println!("failed to write reply to socket; err = {:?}", e); +// return; +// } +// } + } + + pub fn start_listening(&self) -> Result<(), Box> { // 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(); // Spawn the root task rt.block_on(async { - let my_address = "127.0.0.1:8081"; - let mut listener = tokio::net::TcpListener::bind(my_address).await?; - - println!("Starting Nym store-and-forward Provider on address {:?}", my_address); - println!("Waiting for input..."); + let mut listener = tokio::net::TcpListener::bind(self.network_address).await?; + let processing_data = ProcessingData::new(self.secret_key, self.store_dir.clone()).add_arc_rwlock(); loop { - let (mut inbound, _) = listener.accept().await?; - + let (socket, _) = listener.accept().await?; + let thread_processing_data = processing_data.clone(); tokio::spawn(async move { - let mut buf = [0; sphinx::PACKET_SIZE]; - - loop { - match inbound.read(&mut buf).await { - Ok(length) if length == 0 => - { - println!("Remote connection closed."); - return; - } - Ok(_) => { - let packet = SphinxPacket::from_bytes(buf.to_vec()).unwrap(); - let payload = match packet.process(Default::default()) { - ProcessedPacket::ProcessedPacketFinalHop(_, _, payload) => Some(payload), - _ => None, - }.unwrap(); - let message = payload.get_content(); - } - Err(e) => { - println!("failed to read from socket; err = {:?}", e); - return; - } - }; - } + ServiceProvider::process_socket_connection(socket, thread_processing_data).await }); } + +// async move { +// let mut incoming = listener.incoming(); +// +// while let Some(conn) = incoming.next().await { +// match conn { +// Err(e) => eprintln!("accept failed with error: {:?}", e), +// Ok(socket) => { +// let foomp2 = processing_data_foomp.clone(); +// tokio::spawn(async move { +// ServiceProvider::process_socket_connection(socket, foomp2).await +// +//// ServiceProvider::process_socket_fixture(socket).await +// }); +// } +// } +// } +// +// }.await; +// +// println!("Server went kaput"); +// Ok(()) + + +//// let processing_data = Arc::new(RwLock::new(ProcessingData::new(self.secret_key, self.store_dir.clone()))); +// let processing_data = Arc::new(RwLock::new((ProcessingData::new(self.secret_key)))); +// +// loop { +// let (socket, _) = listener.accept().await?; +// +//// tokio:: +// tokio::spawn(async move { +//// processing_data.read(); +//// let foo = processing_data.clone(); +//// let foo = ProcessingData::new(self.secret_key.clone()); +// // Process each socket concurrently. +//// ServiceProvider::process_socket_connection(socket, processing_data).await +// }); +// } }) } - } + + + From 6fdc4a643453b934ee6a6ffb919153f2c7f1d6be Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 10 Dec 2019 14:25:30 +0000 Subject: [PATCH 4/9] Calling store_processed_data after processing each sphinx packet --- src/provider/mod.rs | 78 +++++++++++++++++++++++++-------------------- 1 file changed, 43 insertions(+), 35 deletions(-) diff --git a/src/provider/mod.rs b/src/provider/mod.rs index 5e37fd91bb..6ccc9ebcad 100644 --- a/src/provider/mod.rs +++ b/src/provider/mod.rs @@ -1,7 +1,7 @@ use std::borrow::Borrow; use std::cell::RefCell; use std::net::SocketAddr; -use std::path::{PathBuf, Path}; +use std::path::{Path, PathBuf}; use std::pin::Pin; use std::sync::Arc; use std::sync::RwLock; @@ -39,14 +39,14 @@ impl From for MixProcessingError { #[derive(Debug, Clone)] struct ProcessingData { secret_key: Scalar, -store_dir: PathBuf, + store_dir: PathBuf, } impl ProcessingData { - fn new(secret_key: Scalar, store_dir: PathBuf) -> Self { + fn new(secret_key: Scalar, store_dir: PathBuf) -> Self { ProcessingData { secret_key, -store_dir, + store_dir, } } @@ -65,9 +65,10 @@ struct StoreData { struct PacketProcessor(()); impl PacketProcessor { - fn process_sphinx_data_packet(packet_data: &[u8], processing_data: &ProcessingData) -> Result { + fn process_sphinx_data_packet(packet_data: &[u8], processing_data: &RwLock) -> Result { let packet = SphinxPacket::from_bytes(packet_data.to_vec())?; - let (client_address, client_surb_id, payload) = match packet.process(processing_data.secret_key) { + let read_processing_data = processing_data.read().unwrap(); + let (client_address, client_surb_id, payload) = match packet.process(read_processing_data.secret_key) { ProcessedPacket::ProcessedPacketFinalHop(client_address, surb_id, payload) => (client_address, surb_id, payload), _ => return Err(MixProcessingError::ReceivedForwardHopError), }; @@ -84,7 +85,8 @@ impl PacketProcessor { }) } - fn store_processed_data(store_data: StoreData, store_dir: &PathBuf) -> Result<(), MixProcessingError> { + fn store_processed_data(store_data: StoreData, store_dir: &Path) -> Result<(), MixProcessingError> { + println!("going to store: {:?} in base dir: {:?}", store_data.message, store_dir); Ok(()) } } @@ -107,34 +109,40 @@ impl ServiceProvider { async fn process_socket_connection(mut socket: tokio::net::TcpStream, processing_data: Arc>) { - println!("look, we can read shared data here! {:?}", processing_data.read().unwrap()); - // NOTE: processing_data is copied here!! -// let mut buf = [0u8; sphinx::PACKET_SIZE]; -// -// // In a loop, read data from the socket and write the data back. -// loop { -// match socket.read(&mut buf).await { -// // socket closed -// Ok(n) if n == 0 => { -// println!("Remote connection closed."); -// return; -// } -// Ok(_) => { -//// let store_data = PacketProcessor::process_sphinx_data_packet(buf.as_ref(), -//// processing_data.read().unwrap().borrow()).unwrap(); -// } -// Err(e) => { -// println!("failed to read from socket; err = {:?}", e); -// return; -// } -// }; -// -// // Write the some data back -// if let Err(e) = socket.write_all(b"foomp").await { -// println!("failed to write reply to socket; err = {:?}", e); -// return; -// } -// } + let mut buf = [0u8; sphinx::PACKET_SIZE]; + + // In a loop, read data from the socket and write the data back. + loop { + match socket.read(&mut buf).await { + // socket closed + Ok(n) if n == 0 => { + println!("Remote connection closed."); + return; + } + Ok(_) => { + let store_data = match PacketProcessor::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; + } + }; + PacketProcessor::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) + }); + } + Err(e) => { + eprintln!("failed to read from socket; err = {:?}", e); + return; + } + }; + + // Write the some data back + if let Err(e) = socket.write_all(b"foomp").await { + eprintln!("failed to write reply to socket; err = {:?}", e); + return; + } + } } From 7a3f16548b44ed0476ba35c29df85541a9ad9519 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 10 Dec 2019 14:35:44 +0000 Subject: [PATCH 5/9] Creating full store path --- Cargo.lock | 7 +++++++ Cargo.toml | 9 +++++---- src/provider/mod.rs | 4 +++- 3 files changed, 15 insertions(+), 5 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 510d013640..fb1a172310 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -348,6 +348,11 @@ dependencies = [ "libc 0.2.65 (registry+https://github.com/rust-lang/crates.io-index)", ] +[[package]] +name = "hex" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" + [[package]] name = "hkdf" version = "0.8.0" @@ -507,6 +512,7 @@ dependencies = [ "clap 2.33.0 (registry+https://github.com/rust-lang/crates.io-index)", "curve25519-dalek 1.2.3 (registry+https://github.com/rust-lang/crates.io-index)", "futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "hex 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)", "sphinx 0.1.0", "tokio 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)", ] @@ -880,6 +886,7 @@ dependencies = [ "checksum generic-array 0.12.3 (registry+https://github.com/rust-lang/crates.io-index)" = "c68f0274ae0e023facc3c97b2e00f076be70e254bc851d972503b328db79b2ec" "checksum getrandom 0.1.13 (registry+https://github.com/rust-lang/crates.io-index)" = "e7db7ca94ed4cd01190ceee0d8a8052f08a247aa1b469a7f68c6a3b71afcf407" "checksum hermit-abi 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "307c3c9f937f38e3534b1d6447ecf090cafcc9744e4a6360e8b037b2cf5af120" +"checksum hex 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)" = "023b39be39e3a2da62a94feb433e91e8bcd37676fbc8bea371daf52b7a769a3e" "checksum hkdf 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)" = "3fa08a006102488bd9cd5b8013aabe84955cf5ae22e304c2caf655b633aefae3" "checksum hmac 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)" = "5dcb5e64cda4c23119ab41ba960d1e170a774c8e4b9d9e6a9bc18aabf5e59695" "checksum iovec 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)" = "b2b3ea6ff95e175473f8ffe6a7eb7c00d054240321b84c57051175fe3c1e075e" diff --git a/Cargo.toml b/Cargo.toml index 06dc593285..36ad82e260 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,8 +7,9 @@ edition = "2018" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -sphinx = { path = "../sphinx" } -tokio = { version = "0.2.4", features = ["full"] } -futures = "0.3.1" +clap = "2.33.0" curve25519-dalek = "1.2.3" -clap = "2.33.0" \ No newline at end of file +hex = "0.4.0" +futures = "0.3.1" +sphinx = { path = "../sphinx" } +tokio = { version = "0.2.4", features = ["full"] } \ No newline at end of file diff --git a/src/provider/mod.rs b/src/provider/mod.rs index 6ccc9ebcad..cf2d386545 100644 --- a/src/provider/mod.rs +++ b/src/provider/mod.rs @@ -86,7 +86,9 @@ impl PacketProcessor { } fn store_processed_data(store_data: StoreData, store_dir: &Path) -> Result<(), MixProcessingError> { - println!("going to store: {:?} in base dir: {:?}", store_data.message, store_dir); + let client_dir_name = hex::encode(store_data.client_address); + let full_store_path = store_dir.join(client_dir_name); + println!("going to store: {:?} in dir: {:?}", store_data.message, full_store_path); Ok(()) } } From fd2f565af20879f302bdabb8fe1f4df91a6d434a Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 10 Dec 2019 15:28:51 +0000 Subject: [PATCH 6/9] Provider storing received messages --- Cargo.lock | 1 + Cargo.toml | 1 + .../yzbqQEtPyDEl8jYt | 1 + src/provider/mod.rs | 40 +++++++++++++++++-- 4 files changed, 40 insertions(+), 3 deletions(-) create mode 100644 foomp/0000000000000000000000000000000000000000000000000000000000000000/yzbqQEtPyDEl8jYt diff --git a/Cargo.lock b/Cargo.lock index fb1a172310..58ab40b76e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -513,6 +513,7 @@ dependencies = [ "curve25519-dalek 1.2.3 (registry+https://github.com/rust-lang/crates.io-index)", "futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", "hex 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)", + "rand 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)", "sphinx 0.1.0", "tokio 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)", ] diff --git a/Cargo.toml b/Cargo.toml index 36ad82e260..b62645736e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,5 +11,6 @@ clap = "2.33.0" curve25519-dalek = "1.2.3" hex = "0.4.0" futures = "0.3.1" +rand = "0.7.2" sphinx = { path = "../sphinx" } tokio = { version = "0.2.4", features = ["full"] } \ No newline at end of file diff --git a/foomp/0000000000000000000000000000000000000000000000000000000000000000/yzbqQEtPyDEl8jYt b/foomp/0000000000000000000000000000000000000000000000000000000000000000/yzbqQEtPyDEl8jYt new file mode 100644 index 0000000000..797570b950 --- /dev/null +++ b/foomp/0000000000000000000000000000000000000000000000000000000000000000/yzbqQEtPyDEl8jYt @@ -0,0 +1 @@ +Hello, Sphinx 0 \ No newline at end of file diff --git a/src/provider/mod.rs b/src/provider/mod.rs index cf2d386545..ceb3451726 100644 --- a/src/provider/mod.rs +++ b/src/provider/mod.rs @@ -12,6 +12,13 @@ use sphinx::{ProcessedPacket, SphinxPacket}; use sphinx::route::{DestinationAddressBytes, SURBIdentifier}; use tokio::prelude::*; use tokio::runtime::Runtime; +use rand::Rng; +use std::fs::File; +use std::io::Write; + +// TODO: if we ever create config file, this should go there +const STORED_MESSAGE_FILENAME_LENGTH: usize = 16; + // TODO: this will probably need to be moved elsewhere I imagine // DUPLICATE WITH MIXNODE CODE!!! @@ -21,7 +28,7 @@ pub enum MixProcessingError { ReceivedForwardHopError, InvalidPayload, NonMatchingRecipient, - StoreFailure, + FileIOFailure, } impl From for MixProcessingError { @@ -33,6 +40,15 @@ impl From for MixProcessingError { } } +impl From for MixProcessingError { + fn from(_: std::io::Error) -> Self { + use MixProcessingError::*; + + FileIOFailure + } +} + + // ProcessingData defines all data required to correctly unwrap sphinx packets // Do note that we're copying this struct around and hence the secret_key. // It might, or might not be, what we want @@ -85,10 +101,24 @@ impl PacketProcessor { }) } + fn generate_random_file_name() -> String { + rand::thread_rng().sample_iter(&rand::distributions::Alphanumeric).take(STORED_MESSAGE_FILENAME_LENGTH).collect::() + } + fn store_processed_data(store_data: StoreData, store_dir: &Path) -> Result<(), MixProcessingError> { let client_dir_name = hex::encode(store_data.client_address); - let full_store_path = store_dir.join(client_dir_name); - println!("going to store: {:?} in dir: {:?}", store_data.message, full_store_path); + let full_store_dir = store_dir.join(client_dir_name); + let full_store_path = full_store_dir.join(PacketProcessor::generate_random_file_name()); + println!("going to store: {:?} in file: {:?}", store_data.message, full_store_path); + + // 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 + std::fs::create_dir_all(full_store_dir)?; + let mut file = File::create(full_store_path)?; + file.write_all(store_data.message.as_ref())?; + + Ok(()) } } @@ -111,6 +141,10 @@ impl ServiceProvider { async fn process_socket_connection(mut socket: tokio::net::TcpStream, processing_data: Arc>) { + // TODO: we will actually need to distinguish multiple types of requests here; + // either from mixnodes to store final hop information + // or from clients to pull messages + let mut buf = [0u8; sphinx::PACKET_SIZE]; // In a loop, read data from the socket and write the data back. From bf360473d53a533669b9ad2ea6896c3eb9796308 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 10 Dec 2019 15:29:08 +0000 Subject: [PATCH 7/9] Removed accidentally added tmp store directory --- .../yzbqQEtPyDEl8jYt | 1 - 1 file changed, 1 deletion(-) delete mode 100644 foomp/0000000000000000000000000000000000000000000000000000000000000000/yzbqQEtPyDEl8jYt diff --git a/foomp/0000000000000000000000000000000000000000000000000000000000000000/yzbqQEtPyDEl8jYt b/foomp/0000000000000000000000000000000000000000000000000000000000000000/yzbqQEtPyDEl8jYt deleted file mode 100644 index 797570b950..0000000000 --- a/foomp/0000000000000000000000000000000000000000000000000000000000000000/yzbqQEtPyDEl8jYt +++ /dev/null @@ -1 +0,0 @@ -Hello, Sphinx 0 \ No newline at end of file From d08a9965c8164d80a6e717e0ab8b55bd797a3038 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 10 Dec 2019 15:34:58 +0000 Subject: [PATCH 8/9] Removed commented code with an experiment on usage of tokio tcp listener --- src/provider/mod.rs | 39 --------------------------------------- 1 file changed, 39 deletions(-) diff --git a/src/provider/mod.rs b/src/provider/mod.rs index ceb3451726..e15fe12684 100644 --- a/src/provider/mod.rs +++ b/src/provider/mod.rs @@ -200,45 +200,6 @@ impl ServiceProvider { ServiceProvider::process_socket_connection(socket, thread_processing_data).await }); } - -// async move { -// let mut incoming = listener.incoming(); -// -// while let Some(conn) = incoming.next().await { -// match conn { -// Err(e) => eprintln!("accept failed with error: {:?}", e), -// Ok(socket) => { -// let foomp2 = processing_data_foomp.clone(); -// tokio::spawn(async move { -// ServiceProvider::process_socket_connection(socket, foomp2).await -// -//// ServiceProvider::process_socket_fixture(socket).await -// }); -// } -// } -// } -// -// }.await; -// -// println!("Server went kaput"); -// Ok(()) - - -//// let processing_data = Arc::new(RwLock::new(ProcessingData::new(self.secret_key, self.store_dir.clone()))); -// let processing_data = Arc::new(RwLock::new((ProcessingData::new(self.secret_key)))); -// -// loop { -// let (socket, _) = listener.accept().await?; -// -//// tokio:: -// tokio::spawn(async move { -//// processing_data.read(); -//// let foo = processing_data.clone(); -//// let foo = ProcessingData::new(self.secret_key.clone()); -// // Process each socket concurrently. -//// ServiceProvider::process_socket_connection(socket, processing_data).await -// }); -// } }) } } From 528ba874929d91ae21fd6b403d9f0a42c3697141 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 10 Dec 2019 15:36:14 +0000 Subject: [PATCH 9/9] Removed unused imports --- src/provider/mod.rs | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/src/provider/mod.rs b/src/provider/mod.rs index e15fe12684..fcc5ba3c69 100644 --- a/src/provider/mod.rs +++ b/src/provider/mod.rs @@ -1,20 +1,16 @@ -use std::borrow::Borrow; -use std::cell::RefCell; +use std::fs::File; +use std::io::Write; use std::net::SocketAddr; use std::path::{Path, PathBuf}; -use std::pin::Pin; use std::sync::Arc; use std::sync::RwLock; use curve25519_dalek::scalar::Scalar; -use futures::StreamExt; +use rand::Rng; use sphinx::{ProcessedPacket, SphinxPacket}; use sphinx::route::{DestinationAddressBytes, SURBIdentifier}; use tokio::prelude::*; use tokio::runtime::Runtime; -use rand::Rng; -use std::fs::File; -use std::io::Write; // TODO: if we ever create config file, this should go there const STORED_MESSAGE_FILENAME_LENGTH: usize = 16;