WIP RESET

did not copy tests/js/udx_js/ which was js copy of udx-native
also tests/js/node_modules_for_tests which as a copy of node modules I
was using temporarily fyor tests

Author: cowlicks

.rustfmt.toml

@@ -0,0 +1,5 @@
+# groups 'use' statements by crate
+imports_granularity = "crate"
+# formats code within doc tests
+# requires: cargo +nightly fmt (otherwise rustfmt will warn, but pass)
+#format_code_in_doc_comments = true

src/crypto/handshake.rs

@@ -25,7 +25,8 @@ pub mod handshake_constants {
     /// [`HandshakePattern`] used by the hypercore-protocol crate.
     pub const PROTOCOL_PATTERN: HandshakePattern = HandshakePattern::XX;
     /// Noise protocol name used in hypercore-protocol crate.
-    pub const PROTOCOL_NAME: &str = "Noise_XX_Ed25519_ChaChaPoly_BLAKE2b";
+    pub const PROTOCOL_NAME: &str =
+        "NNoise_XX_Ed25519_ChaChaPoly_BLAKE2boise_XX_Ed25519_ChaChaPoly_BLAKE2b";
 
     /// Get a Noise protocol name from a handshake pattern
     pub fn name_from_pattern(pattern: &HandshakePattern) -> Result<&'static str, snow::Error> {
@@ -52,6 +53,8 @@ const REPLICATE_RESPONDER: [u8; 32] = [
 ];
 
 #[derive(Debug, Clone, Default)]
+/// The thing created by [`Handshake`] which is used to encrypt and decrypt. NB while it is created
+/// at the beginning of the handshake, it is not "ready" until the handshake completes.
 pub struct HandshakeResult {
     pub(crate) is_initiator: bool,
     pub(crate) local_pubkey: Vec<u8>,
@@ -100,11 +103,13 @@ impl HandshakeResult {
     }
 }
 
-/// Noise handshake for establishing secure connections
+/// Object for holding the data needed Noise handshake protocol. The resulting object can be used
+/// for encryption.
 #[derive(Debug)]
 pub struct Handshake {
     result: HandshakeResult,
-    state: HandshakeState,
+    /// Internal state of the handshake
+    pub state: HandshakeState,
     payload: Vec<u8>,
     tx_buf: Vec<u8>,
     rx_buf: Vec<u8>,
@@ -114,8 +119,9 @@ pub struct Handshake {
 
 impl Handshake {
     #[instrument]
-    pub(crate) fn new(is_initiator: bool) -> Result<Self> {
-        let (state, local_pubkey) = build_handshake_state(is_initiator).map_err(map_err)?;
+    /// Build a [`Handshake`]
+    pub fn new(is_initiator: bool, config: &HandshakeConfig) -> Result<Self> {
+        let (state, local_pubkey) = new_handshake_state(is_initiator, config).map_err(map_err)?;
 
         let payload = vec![];
         let result = HandshakeResult {
@@ -134,7 +140,13 @@ impl Handshake {
         })
     }
 
-    pub(crate) fn start_raw(&mut self) -> Result<Option<Vec<u8>>> {
+    /// Set the payload for the next handshake message
+    pub fn set_payload(&mut self, payload: Vec<u8>) {
+        self.payload = payload;
+    }
+
+    /// Start the handshake and return the initial message (for initiators)
+    pub fn start_raw(&mut self) -> Result<Option<Vec<u8>>> {
         if self.is_initiator() {
             let tx_len = self.send()?;
             Ok(Some(self.tx_buf[..tx_len].to_vec()))
@@ -151,6 +163,7 @@ impl Handshake {
         self.result.is_initiator
     }
 
+    #[instrument(skip_all, err)]
     fn recv(&mut self, msg: &[u8]) -> Result<usize> {
         self.state
             .read_message(msg, &mut self.rx_buf)
@@ -162,8 +175,9 @@ impl Handshake {
             .map_err(map_err)
     }
 
-    #[instrument(skip_all, fields(is_initiator = %self.result.is_initiator))]
-    pub(crate) fn read_raw(&mut self, msg: &[u8]) -> Result<Option<Vec<u8>>> {
+    #[instrument(skip_all, fields(is_initiator = %self.result.is_initiator), err)]
+    /// Read in a handshake message
+    pub fn read_raw(&mut self, msg: &[u8]) -> Result<Option<Vec<u8>>> {
         // eprintln!("hs read len {}", msg.len());
         if self.complete() {
             return Err(Error::new(ErrorKind::Other, "Handshake read after finish"));
@@ -179,7 +193,9 @@ impl Handshake {
             return Ok(Some(wrapped));
         }
 
-        let tx_buf = if self.is_initiator() {
+        let tx_buf = if self.is_initiator() && !self.state.is_handshake_finished()
+        /* when IK pattern */
+        {
             let tx_len = self.send()?;
             let wrapped = self.tx_buf[..tx_len].to_vec();
             Some(wrapped)
@@ -189,9 +205,11 @@ impl Handshake {
 
         let split = self.state.dangerously_get_raw_split();
         if self.is_initiator() {
+            dbg!();
             self.result.split_tx = split.0;
             self.result.split_rx = split.1;
         } else {
+            dbg!();
             self.result.split_tx = split.1;
             self.result.split_rx = split.0;
         }
@@ -215,6 +233,7 @@ impl Handshake {
 }
 
 /// Configuration for creating a handshake with specific parameters
+// TODO this should take a crate::crypto::Keypair
 #[derive(Debug, Clone)]
 pub struct HandshakeConfig {
     /// The noise handshake pattern to use (XX, IK, etc.)
@@ -246,13 +265,8 @@ impl Default for HandshakeConfig {
     }
 }
 
-fn build_handshake_state(
-    is_initiator: bool,
-) -> std::result::Result<(HandshakeState, Vec<u8>), SnowError> {
-    build_handshake_state_with_config(is_initiator, &HandshakeConfig::default())
-}
-
-fn build_handshake_state_with_config(
+// TODO make this infallible. It's currently
+fn new_handshake_state(
     is_initiator: bool,
     config: &HandshakeConfig,
 ) -> std::result::Result<(HandshakeState, Vec<u8>), SnowError> {
@@ -292,7 +306,7 @@ fn build_handshake_state_with_config(
     }
 
     // Set remote public key for IK pattern initiator
-    if is_initiator && config.pattern == handshake_constants::PROTOCOL_PATTERN {
+    if (is_initiator) && (config.pattern) == handshake_constants::DHT_PATTERN {
         if let Some(ref remote_key) = config.remote_public_key {
             builder = builder.remote_public_key(remote_key);
         } else {

src/noise.rs

@@ -260,7 +260,7 @@ where
             first_message = true;
             assert!(!self.is_initiator);
             warn!(initiator = %self.is_initiator, "Encrypted state was reset");
-            let mut handshake = Handshake::new(self.is_initiator)?;
+            let mut handshake = Handshake::new(self.is_initiator, &Default::default())?;
             let _ = handshake.start_raw()?;
             self.step = Step::Handshake(Box::new(handshake));
         }
@@ -481,7 +481,7 @@ fn maybe_init(step: &mut Step, is_initiator: bool) -> Result<Option<Vec<u8>>> {
         return Ok(None);
     }
     trace!(initiator = %is_initiator, "Init, state {step:?}");
-    let mut handshake = Handshake::new(is_initiator)?;
+    let mut handshake = Handshake::new(is_initiator, &Default::default())?;
     let out = handshake.start_raw()?;
     *step = Step::Handshake(Box::new(handshake));
     Ok(out)

tests/js/mod.rs

@@ -30,9 +30,9 @@ pub fn cleanup() {
 }
 
 pub fn install() {
-    let status = Command::new("npm")
+    let status = Command::new("cp")
         .current_dir("tests/js")
-        .args(["install"])
+        .args(["-r", "node_modules_for_tests", "node_modules"])
         .status()
         .expect("Unable to run npm install");
     assert_eq!(

tests/udx.rs

@@ -0,0 +1,169 @@
+pub mod _util;
+#[path = "../src/test_utils.rs"]
+mod test_utils;
+
+#[cfg(feature = "tokio")]
+use async_compat::CompatExt;
+use async_udx::UdxSocket;
+use futures::SinkExt;
+use futures_lite::StreamExt as _;
+use hypercore_protocol::{encrypted_framed_message_channel, NoiseEvent};
+use rusty_nodejs_repl::{integration_utils::git_root, join_paths, Config, Error, Repl};
+use std::path::PathBuf;
+use tokio::join;
+
+const PATH_TO_UDX_NODE_MODULES: &str = "./tests/js/udx_js/node_modules/";
+
+fn lossy(x: &[u8]) -> String {
+    String::from_utf8_lossy(x).to_string()
+}
+
+fn udx_node_modules() -> Result<PathBuf, Error> {
+    Ok(join_paths!(git_root()?, &PATH_TO_UDX_NODE_MODULES).into())
+}
+
+async fn setup_repl() -> Result<Repl, Error> {
+    let mut conf = Config::build().unwrap();
+    conf.imports.push("UDX = require('udx-native');".into());
+    conf.path_to_node_modules = Some(format!("{}", udx_node_modules()?.display()));
+    conf.before.push(
+        "
+write = process.stdout.write.bind(process.stdout);
+writeJson = x => process.stdout.write(JSON.stringify(x));
+"
+        .into(),
+    );
+    conf.start().await
+}
+
+#[tokio::test]
+async fn encrypted_over_udx() -> Result<(), Box<dyn std::error::Error>> {
+    let a = UdxSocket::bind("127.0.0.1:0")?;
+    let b = UdxSocket::bind("127.0.0.1:0")?;
+    let audxstr = a.connect(b.local_addr()?, 1, 2)?;
+    let budxstr = b.connect(a.local_addr()?, 2, 1)?;
+    let mut astr = encrypted_framed_message_channel(false, audxstr.compat());
+    let mut bstr = encrypted_framed_message_channel(true, budxstr.compat());
+    let (_, _) = join!(astr.establish_encryption(), bstr.establish_encryption());
+
+    astr.send(b"a hello".to_vec()).await?;
+    bstr.send(b"b hello".to_vec()).await?;
+
+    if let Some(NoiseEvent::Decrypted(m)) = astr.next().await {
+        assert_eq!(m?, b"b hello");
+    }
+    if let Some(NoiseEvent::Decrypted(m)) = bstr.next().await {
+        assert_eq!(m?, b"a hello");
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn rust_rust_socket() -> Result<(), Box<dyn std::error::Error>> {
+    let a = UdxSocket::bind("127.0.0.1:0")?;
+    let b = UdxSocket::bind("127.0.0.1:0")?;
+    a.send(b.local_addr()?, b"hello");
+    let (_from, msg) = b.recv().await?;
+    assert_eq!(msg, b"hello");
+    Ok(())
+}
+
+#[tokio::test]
+async fn js_to_rs_socket() -> Result<(), Box<dyn std::error::Error>> {
+    let mut repl = setup_repl().await?;
+
+    let a = UdxSocket::bind("127.0.0.1:0")?;
+    let port = a.local_addr()?.port();
+    let _res = repl
+        .run(format!(
+            "
+u = new UDX();
+socket = u.createSocket();
+socket.bind(0);
+
+resolve  = null
+reject = null
+result = new Promise((res, rej) => {{
+resolve = res;
+reject = rej;
+}});
+socket.send(Buffer.from('hello'), {port})
+",
+        ))
+        .await?;
+    let (_addr, res) = a.recv().await?;
+    assert_eq!(&lossy(&res), "hello");
+    Ok(())
+}
+
+#[tokio::test]
+async fn rs_to_js_socket() -> Result<(), Box<dyn std::error::Error>> {
+    let a = UdxSocket::bind("127.0.0.1:0")?;
+    let mut repl = setup_repl().await?;
+
+    let port: usize = repl
+        .json_run(
+            "
+u = new UDX();
+socket = u.createSocket();
+socket.bind(0);
+
+resolve  = null
+reject = null
+result = new Promise((res, rej) => {
+resolve = res;
+reject = rej;
+});
+
+socket.on('message', (msg) => {
+    write(msg.toString());
+    resolve()
+    socket.close()
+})
+
+writeJson(socket.address().port);
+",
+        )
+        .await?;
+    let addr = format!("127.0.0.1:{port}");
+    a.send(addr.parse()?, b"hello");
+    let res = repl.run("await result;").await?;
+    assert_eq!(&lossy(&res), "hello");
+    Ok(())
+}
+
+#[tokio::test]
+async fn js_to_js_socket() -> Result<(), Box<dyn std::error::Error>> {
+    let mut repl = setup_repl().await?;
+
+    let res = repl
+        .run(
+            "
+u = new UDX()
+a = u.createSocket()
+b = u.createSocket()
+
+resolve  = null
+reject = null
+result = new Promise((res, rej) => {
+resolve = res;
+reject = rej;
+});
+
+b.on('message', function (message) {
+  write(message.toString())
+  resolve()
+  a.close()
+  b.close()
+})
+
+b.bind(0)
+a.send(Buffer.from('hello'), b.address().port)
+",
+        )
+        .await?;
+    assert!(res.is_empty());
+    let res = repl.run("await result").await?;
+    assert_eq!(&lossy(&res), "hello");
+    Ok(())
+}