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

Revert "Starting to unify the slightly different interfaces of NymClient"

Browse Source 
This reverts commit 4d444ceb07.
This commit is contained in:
Dave Hrycyszyn
2020-08-15 12:06:44 +01:00
parent 4d444ceb07
commit eec6034684
4 changed files with 51 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/client-core/src/client/received_buffer.rs
+1 -1
View File
@@ -150,7 +150,7 @@ impl ReceivedMessagesBuffer {
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!") // TODO: WTF is happening here?
panic!("trying to disconnect non-existent sender!")
}
guard.message_sender = None;
}
clients/native/src/client/mod.rs
+13 -37
View File
@@ -44,7 +44,6 @@ use nymsphinx::addressing::clients::Recipient;
use nymsphinx::addressing::nodes::NodeIdentity;
use nymsphinx::anonymous_replies::ReplySURB;
use nymsphinx::receiver::ReconstructedMessage;
use tokio::runtime;
use tokio::runtime::Runtime;
pub struct NymClient {
@@ -259,8 +258,18 @@ impl NymClient {
MixTrafficController::new(mix_rx, gateway_client).start(self.runtime.handle());
}
fn start_listening(&self, listener: Box<dyn ClientListener>) {
listener.listen(self.config.get_listening_port(), self.runtime.handle());
fn start_websocket_listener(
&self,
buffer_requester: ReceivedBufferRequestSender,
msg_input: InputMessageSender,
) {
info!("Starting websocket listener...");
let websocket_handler =
websocket::Handler::new(msg_input, buffer_requester, self.as_mix_recipient());
websocket::Listener::new(self.config.get_listening_port())
.start(self.runtime.handle(), websocket_handler);
}
/// EXPERIMENTAL DIRECT RUST API
@@ -371,16 +380,10 @@ impl NymClient {
sphinx_message_sender.clone(),
);
self.start_cover_traffic_stream(shared_topology_accessor, sphinx_message_sender);
let recipient = self.as_mix_recipient();
let websocket = Box::new(WebsocketListener::new(
received_buffer_request_sender.clone(),
input_sender.clone(),
recipient,
));
match self.config.get_socket_type() {
SocketType::WebSocket => {
self.start_listening(websocket);
self.start_websocket_listener(received_buffer_request_sender, input_sender)
}
SocketType::None => {
// if we did not start the socket, it means we're running (supposedly) in the native mode
@@ -414,30 +417,3 @@ impl NymClient {
);
}
}
pub(crate) trait ClientListener {
fn listen(&self, port: u16, runtime_handle: &runtime::Handle);
}
struct WebsocketListener {
handler: websocket::Handler,
}
impl ClientListener for WebsocketListener {
fn listen(&self, port: u16, runtime_handle: &runtime::Handle) {
info!("Starting websocket listener...");
websocket::Listener::new(port).start(runtime_handle, self.handler.clone());
}
}
impl WebsocketListener {
fn new(
received_buffer_request_sender: ReceivedBufferRequestSender,
input_sender: InputMessageSender,
recipient: Recipient,
) -> WebsocketListener {
let handler =
websocket::Handler::new(input_sender, received_buffer_request_sender, recipient);
WebsocketListener { handler }
}
}
clients/socks5/src/client/mod.rs
+30 -65
View File
@@ -46,7 +46,7 @@ use nymsphinx::addressing::clients::{ClientEncryptionKey, ClientIdentity, Recipi
use nymsphinx::addressing::nodes::NodeIdentity;
use nymsphinx::anonymous_replies::ReplySURB;
use nymsphinx::receiver::ReconstructedMessage;
use tokio::runtime::{self, Runtime};
use tokio::runtime::Runtime;
pub struct NymClient {
/// Client configuration options, including, among other things, packet sending rates,
@@ -260,9 +260,35 @@ impl NymClient {
MixTrafficController::new(mix_rx, gateway_client).start(self.runtime.handle());
}
fn start_listening(&self, listener: Box<dyn ClientListener>) {
fn start_socks5_listener(
&self,
buffer_requester: ReceivedBufferRequestSender,
msg_input: InputMessageSender,
) {
info!("Starting socks5 listener...");
listener.listen(self.config.get_listening_port(), self.runtime.handle());
let mut auth_methods: Vec<u8> = Vec::new();
auth_methods.push(AuthenticationMethods::NoAuth as u8);
let allowed_users: Vec<User> = Vec::new();
let authenticator = Authenticator::new(auth_methods, allowed_users);
let recipient = self.load_socks5_service_provider();
let mut sphinx_socks = SphinxSocksServer::new(1080, "127.0.0.1", authenticator, recipient);
self.runtime
.spawn(async move { sphinx_socks.serve(msg_input, buffer_requester).await });
}
// TODO: make this configurable in the client config file
// TODO: Talk to JS about where I can easily find these.
fn load_socks5_service_provider(&self) -> Recipient {
// load from file here, or better yet, inject it
let identity = "7qBXQor8nHXUXDAUM4aLvqDJ2MECFKn3AJ3brQsu5qz8";
let encryption_key = "4uzn7m3vPEy5MhPHLqzFCJHk2BCnvbCsGdrYzng7jnor";
let gateway_key = "e3vUAo6YhB7zq3GH8B4k3iiGT4H2USjdd5ZMZoUsHdF";
let client_identity = ClientIdentity::from_base58_string(identity).unwrap();
let client_encryption_key =
ClientEncryptionKey::from_base58_string(encryption_key).unwrap();
let gateway = NodeIdentity::from_base58_string(gateway_key).unwrap();
Recipient::new(client_identity, client_encryption_key, gateway)
}
/// EXPERIMENTAL DIRECT RUST API
@@ -374,17 +400,9 @@ impl NymClient {
);
self.start_cover_traffic_stream(shared_topology_accessor, sphinx_message_sender);
let recipient = load_socks5_service_provider();
let socks = Box::new(SocksListener::new(
received_buffer_request_sender.clone(),
input_sender.clone(),
recipient,
));
// let socks = Box::new(listener);
match self.config.get_socket_type() {
SocketType::WebSocket => {
self.start_listening(socks);
self.start_socks5_listener(received_buffer_request_sender, input_sender)
}
SocketType::None => {
// if we did not start the socket, it means we're running (supposedly) in the native mode
@@ -418,56 +436,3 @@ impl NymClient {
);
}
}
// TODO: make this configurable in the client config file
// TODO: Talk to JS about where I can easily find these.
fn load_socks5_service_provider() -> Recipient {
// load from file here, or better yet, inject it
let identity = "7qBXQor8nHXUXDAUM4aLvqDJ2MECFKn3AJ3brQsu5qz8";
let encryption_key = "4uzn7m3vPEy5MhPHLqzFCJHk2BCnvbCsGdrYzng7jnor";
let gateway_key = "e3vUAo6YhB7zq3GH8B4k3iiGT4H2USjdd5ZMZoUsHdF";
let client_identity = ClientIdentity::from_base58_string(identity).unwrap();
let client_encryption_key = ClientEncryptionKey::from_base58_string(encryption_key).unwrap();
let gateway = NodeIdentity::from_base58_string(gateway_key).unwrap();
Recipient::new(client_identity, client_encryption_key, gateway)
}
trait ClientListener {
fn listen(&mut self, port: u16, runtime_handle: &runtime::Handle);
}
struct SocksListener {
server: SphinxSocksServer,
}
impl ClientListener for SocksListener {
fn listen(&mut self, port: u16, runtime_handle: &runtime::Handle) {
info!("Starting socks listener...");
runtime_handle.spawn(async move { self.server.serve().await });
}
}
impl SocksListener {
fn new(
received_buffer_request_sender: ReceivedBufferRequestSender,
input_sender: InputMessageSender,
recipient: Recipient,
) -> SocksListener {
let mut auth_methods: Vec<u8> = Vec::new();
auth_methods.push(AuthenticationMethods::NoAuth as u8);
let allowed_users: Vec<User> = Vec::new();
let authenticator = Authenticator::new(auth_methods, allowed_users);
let server = SphinxSocksServer::new(
1080,
"127.0.0.1",
authenticator,
recipient,
input_sender,
received_buffer_request_sender,
);
SocksListener { server }
}
}
clients/socks5/src/socks/server.rs
+7 -9
View File
@@ -18,8 +18,6 @@ pub struct SphinxSocksServer {
authenticator: Authenticator,
listening_address: SocketAddr,
service_provider: Recipient,
input_sender: InputMessageSender,
buffer_requester: ReceivedBufferRequestSender,
}
impl SphinxSocksServer {
@@ -29,22 +27,22 @@ impl SphinxSocksServer {
ip: &str,
authenticator: Authenticator,
service_provider: Recipient,
input_sender: InputMessageSender,
buffer_requester: ReceivedBufferRequestSender,
) -> Self {
info!("Listening on {}:{}", ip, port);
SphinxSocksServer {
authenticator,
buffer_requester,
listening_address: format!("{}:{}", ip, port).parse().unwrap(),
service_provider,
input_sender,
}
}
/// Set up the listener and initiate connection handling when something
/// connects to the server.
pub(crate) async fn serve(&mut self) -> Result<(), SocksProxyError> {
pub(crate) async fn serve(
&mut self,
input_sender: InputMessageSender,
buffer_requester: ReceivedBufferRequestSender,
) -> Result<(), SocksProxyError> {
info!("Serving Connections...");
let mut listener = TcpListener::bind(self.listening_address).await.unwrap();
@@ -52,7 +50,7 @@ impl SphinxSocksServer {
let active_streams = Arc::new(Mutex::new(HashMap::new()));
let mut mixnet_response_listener =
MixnetResponseListener::new(self.buffer_requester.clone(), Arc::clone(&active_streams));
MixnetResponseListener::new(buffer_requester, Arc::clone(&active_streams));
tokio::spawn(async move {
mixnet_response_listener.run().await;
@@ -64,7 +62,7 @@ impl SphinxSocksServer {
let mut client = SocksClient::new(
stream,
self.authenticator.clone(),
self.input_sender.clone(),
input_sender.clone(),
self.service_provider.clone(),
Arc::clone(&active_streams),
);