parse package, MsgDriver read/write impl, refactor project to workspace, test_client sending Ping msgs

Cargo.toml

@@ -0,0 +1,11 @@
+[workspace]
+
+members = [
+    "signer",
+    "broker",
+    "parser"
+]
+
+exclude = [
+    "sphinx-key",
+]

broker/Cargo.toml

@@ -11,6 +11,7 @@ pretty_env_logger = "0.4.0"
 confy = "0.4.0"
 tokio = "1.18"
 vls-protocol = { path = "../../validating-lightning-signer/vls-protocol" }
+sphinx-key-parser = { path = "../parser" }
 secp256k1 = { version = "0.20", features = ["rand-std", "bitcoin_hashes"] }
 anyhow = {version = "1", features = ["backtrace"]}
 log = "0.4"

broker/src/main.rs

@@ -1,9 +1,7 @@
-mod msg;
-
+use sphinx_key_parser::MsgDriver;
 use librumqttd::{async_locallink::construct_broker, Config};
 use std::thread;
-use tokio::time::{sleep, Duration};
-use vls_protocol::msgs::{self, Message};
+use vls_protocol::msgs;
 use vls_protocol::serde_bolt::WireString;
 use tokio::sync::mpsc;
 
@@ -31,10 +29,9 @@ fn main() {
         let console_task = tokio::spawn(console);
 
         let pub_task = tokio::spawn(async move {
-
             while let Some(_) = msg_rx.recv().await {
                 let sequence = 0;
-                let mut md = msg::MsgDriver::new_empty(); 
+                let mut md = MsgDriver::new_empty(); 
                 msgs::write_serial_request_header(&mut md, sequence, 0).expect("failed to write_serial_request_header");
                 let ping = msgs::Ping {
                     id: 0,

broker/src/test_client.rs

@@ -1,8 +1,11 @@
-use tokio::{task, time};
 
+use sphinx_key_parser::MsgDriver;
+
+use tokio::{task, time};
 use rumqttc::{self, AsyncClient, MqttOptions, QoS, Event, Packet};
 use std::error::Error;
 use std::time::Duration;
+use vls_protocol::msgs;
 
 #[tokio::main(worker_threads = 1)]
 async fn main() -> Result<(), Box<dyn Error>> {
@@ -24,9 +27,16 @@ async fn main() -> Result<(), Box<dyn Error>> {
         // println!("{:?}", event.unwrap());
         if let Event::Incoming(packet) = event.unwrap() {
             if let Packet::Publish(p) = packet {
-                println!("incoming {:?}", p.payload);
+                // println!("incoming {:?}", p.payload);
+                let mut m = MsgDriver::new(p.payload.to_vec());
+                let (sequence, dbid) = msgs::read_serial_request_header(&mut m).expect("read ping header");
+                assert_eq!(dbid, 0);
+                assert_eq!(sequence, 0);
+                let ping: msgs::Ping =
+                    msgs::read_message(&mut m).expect("failed to read ping message");
+                println!("INCOMING: {:?}", ping);
             }
-        } 
+        }
     }
 }
 

parser/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "sphinx-key-parser"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+vls-protocol = { path = "../../validating-lightning-signer/vls-protocol" }
+
+[features]
+default = ["std"]
+std = ["vls-protocol/std"]
+

parser/src/lib.rs

@@ -1,4 +1,6 @@
 use vls_protocol::serde_bolt::{self, Read, Write};
+use std::io;
+use std::cmp::min;
 
 pub struct MsgDriver(Vec<u8>);
 
@@ -9,6 +11,9 @@ impl MsgDriver {
     pub fn new_empty() -> Self {
         Self(Vec::new())
     }
+    pub fn as_ref(&self) -> &Vec<u8> {
+        &self.0
+    }
     pub fn bytes(&self) -> Vec<u8> {
         self.0.clone()
     }
@@ -22,12 +27,21 @@ impl Read for MsgDriver {
         if buf.is_empty() {
             return Ok(0);
         }
-        let len = self.0.len();
-        Ok(len)
+        let (mut content, remaining) = self.0.split_at(
+            min(buf.len(), self.0.len())
+        );
+        let bytes = &mut content;
+        match io::copy(bytes, &mut buf) {
+            Ok(len) => {
+                self.0 = remaining.to_vec();
+                Ok(len as usize)
+            },
+            Err(_) => Ok(0)
+        }
     }
 
     fn peek(&mut self) -> serde_bolt::Result<Option<u8>> {
-        Ok(Some(0))
+        Ok(if let Some(u) = self.0.get(0) { Some(u.clone()) } else { None})
     }
 }
 

signer/.cargo/config.toml

@@ -1,26 +0,0 @@
-[build]
-target = "riscv32imc-esp-espidf"
-
-[target.riscv32imc-esp-espidf]
-linker = "ldproxy"
-
-# Future - necessary for the experimental "native build" of esp-idf-sys with ESP32C3
-# See also https://github.com/ivmarkov/embuild/issues/16
-rustflags = ["-C", "default-linker-libraries"]
-
-# [target.riscv32imc-esp-espidf.rustsecp256k1_v0_5_0]
-# linker = "/Users/evanfeenstra/code/sphinx-key/sphinx-key/signer/.embuild/espressif/tools/riscv32-esp-elf/esp-2021r2-patch3-8.4.0/riscv32-esp-elf/bin/riscv32-esp-elf-gcc"
-# rustc-link-search = ["/Users/evanfeenstra/code/sphinx-key/rust-secp256k1/target/riscv32imc-esp-espidf/debug/libsecp256k1_sys.rlib"]
-# rustc-link-lib = ["rustsecp256k1_v0_5_0"]
-
-[unstable]
-
-build-std = ["std", "panic_abort"]
-#build-std-features = ["panic_immediate_abort"] # Required for older ESP-IDF versions without a realpath implementation
-
-[env]
-# Note: these variables are not used when using pio builder
-# Enables the esp-idf-sys "native" build feature (`cargo build --features native`) to build against ESP-IDF stable (v4.4)
-ESP_IDF_VERSION = { value = "branch:release/v4.4" }
-# Enables the esp-idf-sys "native" build feature (`cargo build --features native`) to build against ESP-IDF master (mainline)
-#ESP_IDF_VERSION = { value = "master" }

signer/Cargo.toml

@@ -3,16 +3,9 @@ name = "sphinx-key-signer"
 version = "0.1.0"
 authors = ["Evan Feenstra <evanfeenstra@gmail.com>"]
 edition = "2018"
-resolver = "2"
-
-[profile.release]
-opt-level = "s"
-
-[profile.dev]
-debug = true # Symbols are nice and they don't increase the size on Flash
-opt-level = "z"
 
 [dependencies]
+sphinx-key-parser = { path = "../parser" }
 vls-protocol-signer = { path = "../../validating-lightning-signer/vls-protocol-signer", default-features = false, features = ["secp-lowmemory", "vls-std"] }
 anyhow = {version = "1", features = ["backtrace"]}
 log = "0.4"

signer/rust-toolchain.toml

@@ -1,3 +0,0 @@
-[toolchain]
-
-channel = "nightly"

signer/sdkconfig.defaults

@@ -1,10 +0,0 @@
-# Rust often needs a bit of an extra main task stack size compared to C (the default is 3K)
-CONFIG_ESP_MAIN_TASK_STACK_SIZE=7000
-
-# Use this to set FreeRTOS kernel tick frequency to 1000 Hz (100 Hz by default).
-# This allows to use 1 ms granuality for thread sleeps (10 ms by default).
-#CONFIG_FREERTOS_HZ=1000
-
-# Workaround for https://github.com/espressif/esp-idf/issues/7631
-CONFIG_MBEDTLS_CERTIFICATE_BUNDLE=n
-CONFIG_MBEDTLS_CERTIFICATE_BUNDLE_DEFAULT_FULL=n

signer/src/lib.rs

@@ -1,14 +1,21 @@
-pub mod msg;
-
+use sphinx_key_parser::MsgDriver;
 use lightning_signer::persist::{DummyPersister, Persist};
 use lightning_signer::Arc;
 use vls_protocol::model::PubKey;
-use vls_protocol::msgs::{self, read_serial_request_header, write_serial_response_header, Message};
+use vls_protocol::msgs;
 use vls_protocol::serde_bolt::WireString;
 use vls_protocol_signer::handler::{Handler, RootHandler};
 use vls_protocol_signer::lightning_signer;
 use vls_protocol_signer::vls_protocol;
 
+pub fn parse_ping(msg_bytes: Vec<u8>) -> msgs::Ping {
+    let mut m = MsgDriver::new(msg_bytes);
+    let (sequence, dbid) = msgs::read_serial_request_header(&mut m).expect("read ping header");
+    let ping: msgs::Ping =
+        msgs::read_message(&mut m).expect("failed to read ping message");
+    ping
+}
+
 pub fn say_hi() {
     let persister: Arc<dyn Persist> = Arc::new(DummyPersister);
 

signer/src/msg.rs

@@ -1,39 +0,0 @@
-
-use anyhow::Result;
-use log::*;
-use vls_protocol_signer::vls_protocol::serde_bolt::{self, Read, Write};
-
-pub struct MsgDriver(Vec<u8>);
-
-impl MsgDriver {
-    pub fn new(raw: Vec<u8>) -> Self {
-        Self(raw)
-    }
-}
-
-impl Read for MsgDriver {
-    type Error = serde_bolt::Error;
-
-    // input: buf to be written. Should already be the right size
-    fn read(&mut self, mut buf: &mut [u8]) -> serde_bolt::Result<usize> {
-        if buf.is_empty() {
-            return Ok(0);
-        }
-        let len = self.0.len();
-        Ok(len)
-    }
-
-    fn peek(&mut self) -> serde_bolt::Result<Option<u8>> {
-        Ok(Some(0))
-    }
-
-}
-
-impl Write for MsgDriver {
-    type Error = serde_bolt::Error;
-
-    fn write_all(&mut self, buf: &[u8]) -> serde_bolt::Result<()> {
-
-        Ok(())
-    }
-}

sphinx-key/src/conn/mqtt.rs

@@ -17,6 +17,8 @@ use std::sync::{Arc, Mutex};
 pub const TOPIC: &str = "sphinx";
 pub const RETURN_TOPIC: &str = "sphinx-return";
 pub const CLIENT_ID: &str = "sphinx-1";
+pub const USERNAME: &str = "sphinx-key";
+pub const PASSWORD: &str = "sphinx-key-pass";
 
 pub fn make_client(broker: &str) -> Result<(
     EspMqttClient<ConnState<MessageImpl, EspError>>, 
@@ -26,6 +28,8 @@ pub fn make_client(broker: &str) -> Result<(
         client_id: Some(CLIENT_ID),
         buffer_size: 2048,
         task_stack: 12288,
+        username: Some(USERNAME),
+        password: Some(PASSWORD),
         // FIXME - mqtts
         // crt_bundle_attach: Some(esp_idf_sys::esp_crt_bundle_attach),
         ..Default::default()

sphinx-key/src/core/events.rs

@@ -1,5 +1,6 @@
 use crate::conn::mqtt::RETURN_TOPIC;
 use crate::periph::led::Led;
+use sphinx_key_signer::parse_ping;
 
 use esp_idf_svc::eventloop::*;
 use embedded_svc::httpd::Result;
@@ -82,16 +83,19 @@ pub fn make_eventloop(client: Arc<Mutex<EspMqttClient<ConnState<MessageImpl, Esp
     let subscription = eventloop.subscribe(move |message: &Message| {
         info!("!!! Got message from the event loop"); //: {:?}", message.0);
         green.blink();
-        let msg_str = message.read_string();
+        let msg_bytes = message.read_bytes();
         // let msg_str = String::from_utf8_lossy(&msg[..]);
         match client.lock() {
-            Ok(mut m_) => if let Err(err) = m_.publish(
-                RETURN_TOPIC,
-                QoS::AtMostOnce,
-                false,
-                format!("The processed message: {}", msg_str).as_bytes(),
-            ) {
-                log::warn!("failed to mqtt publish! {:?}", err);
+            Ok(mut m_) => {
+                let ping = parse_ping(msg_bytes);
+                if let Err(err) = m_.publish(
+                    RETURN_TOPIC,
+                    QoS::AtMostOnce,
+                    false,
+                    format!("Got and parsed the ping!!!").as_bytes(),
+                ) {
+                    log::warn!("failed to mqtt publish! {:?}", err);
+                };
             },
             Err(_) => log::warn!("failed to lock Mutex<Client>")
         };