Implement full DkgSign over StackerDB

libsigner/Cargo.toml

@@ -26,6 +26,7 @@ slog = { version = "2.5.2", features = [ "max_level_trace" ] }
 slog-term = "2.6.0"
 slog-json = { version = "2.3.0", optional = true }
 libc = "0.2"
+tiny_http = "0.12"
 
 [dependencies.serde_json]
 version = "1.0"

libsigner/src/events.rs

@@ -23,13 +23,17 @@ use std::net::SocketAddr;
 use std::net::TcpListener;
 use std::net::TcpStream;
 
-use std::io::Read;
+use std::io::{Read, Write};
 
 use clarity::vm::types::QualifiedContractIdentifier;
 use libstackerdb::StackerDBChunkData;
 
 use serde::{Deserialize, Serialize};
 
+use tiny_http::{
+    Method as HttpMethod, Request as HttpRequest, Response as HttpResponse, Server as HttpServer,
+};
+
 use crate::http::{decode_http_body, decode_http_request};
 use crate::EventError;
 
@@ -106,7 +110,7 @@ pub struct StackerDBEventReceiver {
     /// Address we bind to
     local_addr: Option<SocketAddr>,
     /// server socket that listens for HTTP POSTs from the node
-    sock: Option<TcpListener>,
+    http_server: Option<HttpServer>,
     /// channel into which to write newly-discovered data
     out_channels: Vec<Sender<StackerDBChunksEvent>>,
     /// inter-thread stop variable -- if set to true, then the `main_loop` will exit
@@ -119,7 +123,7 @@ impl StackerDBEventReceiver {
     pub fn new(contract_ids: Vec<QualifiedContractIdentifier>) -> StackerDBEventReceiver {
         let stackerdb_receiver = StackerDBEventReceiver {
             stackerdb_contract_ids: contract_ids,
-            sock: None,
+            http_server: None,
             local_addr: None,
             out_channels: vec![],
             stop_signal: Arc::new(AtomicBool::new(false)),
@@ -128,19 +132,19 @@ impl StackerDBEventReceiver {
     }
 
     /// Do something with the socket
-    pub fn with_socket<F, R>(&mut self, todo: F) -> Result<R, EventError>
+    pub fn with_server<F, R>(&mut self, todo: F) -> Result<R, EventError>
     where
-        F: FnOnce(&mut StackerDBEventReceiver, &mut TcpListener) -> R,
+        F: FnOnce(&mut StackerDBEventReceiver, &mut HttpServer) -> R,
     {
-        let mut sock = if let Some(s) = self.sock.take() {
+        let mut server = if let Some(s) = self.http_server.take() {
             s
         } else {
             return Err(EventError::NotBound);
         };
 
-        let res = todo(self, &mut sock);
+        let res = todo(self, &mut server);
 
-        self.sock = Some(sock);
+        self.http_server = Some(server);
         Ok(res)
     }
 }
@@ -176,44 +180,60 @@ impl EventReceiver for StackerDBEventReceiver {
     /// Returns the address that was bound.
     /// Errors out if bind(2) fails
     fn bind(&mut self, listener: SocketAddr) -> Result<SocketAddr, EventError> {
-        let srv = TcpListener::bind(listener)?;
-        let bound_addr = srv.local_addr()?;
-        self.sock = Some(srv);
-        self.local_addr = Some(bound_addr.clone());
-        Ok(bound_addr)
+        self.http_server = Some(HttpServer::http(listener).expect("failed to start HttpServer"));
+        self.local_addr = Some(listener);
+        Ok(listener)
     }
 
     /// Wait for the node to post something, and then return it.
     /// Errors are recoverable -- the caller should call this method again even if it returns an
     /// error.
     fn next_event(&mut self) -> Result<StackerDBChunksEvent, EventError> {
-        self.with_socket(|event_receiver, server_sock| {
-            let (mut node_sock, _) = server_sock.accept()?;
+        self.with_server(|event_receiver, http_server| {
+            let mut request = http_server.recv()?;
 
             // were we asked to terminate?
             if event_receiver.is_stopped() {
                 return Err(EventError::Terminated);
             }
 
-            let mut buf = vec![];
-            node_sock.read_to_end(&mut buf)?;
-
-            let (verb, path, headers, body_offset) = decode_http_request(&buf)?.destruct();
-            if verb != "POST" {
+            if request.method() != &HttpMethod::Post {
                 return Err(EventError::MalformedRequest(format!(
-                    "Unrecognized verb '{}'",
-                    &verb
+                    "Unrecognized method '{}'",
+                    &request.method(),
                 )));
             }
-            if path != "/stackerdb_chunks" {
-                return Err(EventError::UnrecognizedEvent(path));
+            if request.url() != "/stackerdb_chunks" {
+                let url = request.url().to_string();
+
+                info!(
+                    "[{:?}] next_event got request with unexpected url {}, return OK so other side doesn't keep sending this",
+                    event_receiver.local_addr,
+                    request.url()
+                );
+
+                request
+                    .respond(HttpResponse::empty(200u16))
+                    .expect("response failed");
+                Err(EventError::UnrecognizedEvent(url))
+            } else {
+                let mut body = String::new();
+                request
+                    .as_reader()
+                    .read_to_string(&mut body)
+                    .expect("failed to read body");
+
+                let event: StackerDBChunksEvent =
+                    serde_json::from_slice(body.as_bytes()).map_err(|e| {
+                        EventError::Deserialize(format!("Could not decode body to JSON: {:?}", &e))
+                    })?;
+
+                request
+                    .respond(HttpResponse::empty(200u16))
+                    .expect("response failed");
+
+                Ok(event)
             }
-
-            let body = decode_http_body(&headers, &buf[body_offset..])?;
-            let event: StackerDBChunksEvent = serde_json::from_slice(&body).map_err(|e| {
-                EventError::Deserialize(format!("Could not decode body to JSON: {:?}", &e))
-            })?;
-            Ok(event)
         })?
     }
 

libsigner/src/runloop.rs

@@ -43,15 +43,15 @@ const STDERR: i32 = 2;
 /// Trait describing the needful components of a top-level runloop.
 /// This is where the signer business logic would go.
 /// Implement this, and you get all the multithreaded setup for free.
-pub trait SignerRunLoop<R> {
+pub trait SignerRunLoop<R, CMD: Send> {
     /// Hint to set how long to wait for new events
     fn set_event_timeout(&mut self, timeout: Duration);
     /// Getter for the event poll timeout
     fn get_event_timeout(&self) -> Duration;
     /// Run one pass of the event loop, given new StackerDB events discovered since the last pass.
     /// Returns Some(R) if this is the final pass -- the runloop evaluated to R
     /// Returns None to keep running.
-    fn run_one_pass(&mut self, event: Option<StackerDBChunksEvent>) -> Option<R>;
+    fn run_one_pass(&mut self, event: Option<StackerDBChunksEvent>, cmd: Option<CMD>) -> Option<R>;
 
     /// This is the main loop body for the signer. It continuously receives events from
     /// `event_recv`, polling for up to `self.get_event_timeout()` units of time.  Once it has
@@ -63,10 +63,11 @@ pub trait SignerRunLoop<R> {
     fn main_loop<EVST: EventStopSignaler>(
         &mut self,
         event_recv: Receiver<StackerDBChunksEvent>,
+        command_recv: Receiver<CMD>,
         mut event_stop_signaler: EVST,
     ) -> Option<R> {
         loop {
-            let poll_timeout = self.get_event_timeout();
+            let poll_timeout = Duration::from_millis(128); //self.get_event_timeout();
             let next_event_opt = match event_recv.recv_timeout(poll_timeout) {
                 Ok(event) => Some(event),
                 Err(RecvTimeoutError::Timeout) => None,
@@ -75,8 +76,15 @@ pub trait SignerRunLoop<R> {
                     return None;
                 }
             };
-            if let Some(final_state) = self.run_one_pass(next_event_opt) {
-                // finished!
+            let next_command_opt = match command_recv.recv_timeout(poll_timeout) {
+                Ok(cmd) => Some(cmd),
+                Err(RecvTimeoutError::Timeout) => None,
+                Err(RecvTimeoutError::Disconnected) => {
+                    info!("Command receiver disconnected");
+                    return None;
+                }
+            };
+            if let Some(final_state) = self.run_one_pass(next_event_opt, next_command_opt) {
                 info!("Runloop exit; signaling event-receiver to stop");
                 event_stop_signaler.send();
                 return Some(final_state);
@@ -86,11 +94,13 @@ pub trait SignerRunLoop<R> {
 }
 
 /// The top-level signer implementation
-pub struct Signer<R, SL: SignerRunLoop<R> + Send + Sync, EV: EventReceiver + Send> {
+pub struct Signer<CMD: Send, R, SL: SignerRunLoop<R, CMD> + Send + Sync, EV: EventReceiver + Send> {
     /// the runloop itself
     signer_loop: Option<SL>,
     /// the event receiver to use
     event_receiver: Option<EV>,
+    /// the command receiver to use
+    command_receiver: Option<Receiver<CMD>>,
     /// marker to permit the R type
     _phantom: PhantomData<R>,
 }
@@ -175,15 +185,21 @@ pub fn set_runloop_signal_handler<ST: EventStopSignaler + Send + 'static>(mut st
 }
 
 impl<
+        CMD: Send + 'static,
         R: Send + 'static,
-        SL: SignerRunLoop<R> + Send + Sync + 'static,
+        SL: SignerRunLoop<R, CMD> + Send + Sync + 'static,
         EV: EventReceiver + Send + 'static,
-    > Signer<R, SL, EV>
+    > Signer<CMD, R, SL, EV>
 {
-    pub fn new(runloop: SL, event_receiver: EV) -> Signer<R, SL, EV> {
+    pub fn new(
+        runloop: SL,
+        event_receiver: EV,
+        command_receiver: Receiver<CMD>,
+    ) -> Signer<CMD, R, SL, EV> {
         Signer {
             signer_loop: Some(runloop),
             event_receiver: Some(event_receiver),
+            command_receiver: Some(command_receiver),
             _phantom: PhantomData,
         }
     }
@@ -203,6 +219,10 @@ impl<
             .event_receiver
             .take()
             .ok_or(EventError::AlreadyRunning)?;
+        let command_receiver = self
+            .command_receiver
+            .take()
+            .ok_or(EventError::AlreadyRunning)?;
         let mut signer_loop = self.signer_loop.take().ok_or(EventError::AlreadyRunning)?;
 
         let (event_send, event_recv) = channel();
@@ -226,7 +246,7 @@ impl<
         let runloop_thread = thread::Builder::new()
             .name("signer_runloop".to_string())
             .stack_size(THREAD_STACK_SIZE)
-            .spawn(move || signer_loop.main_loop(event_recv, stop_signaler))
+            .spawn(move || signer_loop.main_loop(event_recv, command_receiver, stop_signaler))
             .map_err(|e| {
                 error!("SignerRunLoop failed to start: {:?}", &e);
                 ret_stop_signaler.send();

libsigner/src/session.rs

@@ -96,9 +96,10 @@ impl StackerDBSession {
     where
         F: FnOnce(&mut StackerDBSession, &mut TcpStream) -> R,
     {
-        if self.sock.is_none() {
-            self.connect_or_reconnect()?;
-        }
+        // TODO: fix this so we can use persistent connection
+        // See https://github.com/stacks-network/stacks-blockchain/issues/3922
+        //if self.sock.is_none() {
+        self.connect_or_reconnect()?;
 
         let mut sock = if let Some(s) = self.sock.take() {
             s

libsigner/src/tests/mod.rs

@@ -19,6 +19,7 @@ mod http;
 use std::io::Write;
 use std::mem;
 use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
+use std::sync::mpsc::{channel, Receiver};
 use std::thread;
 use std::time::Duration;
 
@@ -51,7 +52,11 @@ impl SimpleRunLoop {
     }
 }
 
-impl SignerRunLoop<Vec<StackerDBChunksEvent>> for SimpleRunLoop {
+enum Command {
+    Empty,
+}
+
+impl SignerRunLoop<Vec<StackerDBChunksEvent>, Command> for SimpleRunLoop {
     fn set_event_timeout(&mut self, timeout: Duration) {
         self.poll_timeout = timeout;
     }
@@ -63,6 +68,7 @@ impl SignerRunLoop<Vec<StackerDBChunksEvent>> for SimpleRunLoop {
     fn run_one_pass(
         &mut self,
         event: Option<StackerDBChunksEvent>,
+        _cmd: Option<Command>,
     ) -> Option<Vec<StackerDBChunksEvent>> {
         debug!("Got event: {:?}", &event);
         if let Some(event) = event {
@@ -87,7 +93,8 @@ fn test_simple_signer() {
         "ST2DS4MSWSGJ3W9FBC6BVT0Y92S345HY8N3T6AV7R.hello-world",
     )
     .unwrap()]);
-    let mut signer = Signer::new(SimpleRunLoop::new(5), ev);
+    let (_cmd_send, cmd_recv) = channel();
+    let mut signer = Signer::new(SimpleRunLoop::new(5), ev, cmd_recv);
     let endpoint: SocketAddr = "127.0.0.1:30000".parse().unwrap();
     let thread_endpoint = endpoint.clone();
 

stacks-common/src/util/hash.rs

@@ -429,7 +429,7 @@ where
             row_hashes.reserve(nodes[i].len() / 2);
 
             for j in 0..(nodes[i].len() / 2) {
-                let h = MerkleTree::get_node_hash(&nodes[i][(2 * j)], &nodes[i][2 * j + 1]);
+                let h = MerkleTree::get_node_hash(&nodes[i][2 * j], &nodes[i][2 * j + 1]);
                 row_hashes.push(h);
             }
 

stacks-signer/Cargo.toml

@@ -11,15 +11,25 @@ readme = "README.md"
 resolver = "2"
 edition = "2021"
 
+[lib]
+name = "stacks_signer"
+path = "src/lib.rs"
+
 [[bin]]
 name = "stacks-signer"
 path = "src/main.rs"
 
 [dependencies]
+bincode = "1.3.3"
 clarity = { path = "../clarity" }
 clap = { version = "4.1.1", features = ["derive", "env"] }
+frost-coordinator = { git = "https://github.com/Trust-Machines/stacks-sbtc" }
+frost-signer = { git = "https://github.com/Trust-Machines/stacks-sbtc" }
+hashbrown = "0.14"
 libsigner = { path = "../libsigner" }
 libstackerdb = { path = "../libstackerdb" }
+p256k1 = "5.4"
+rand_core = "0.6"
 serde = "1"
 serde_derive = "1"
 serde_stacker = "0.1"
@@ -29,6 +39,7 @@ slog-term = "2.6.0"
 stacks-common = { path = "../stacks-common" }
 thiserror = "1.0"
 toml = "0.5.6"
+wsts = "2.0"
 
 [dependencies.serde_json]
 version = "1.0"
@@ -37,9 +48,3 @@ features = ["arbitrary_precision", "unbounded_depth"]
 [dependencies.secp256k1]
 version = "0.24.3"
 features = ["serde", "recovery"]
-
-[target.'cfg(all(any(target_arch = "x86_64", target_arch = "x86", target_arch = "aarch64"), not(target_env = "msvc")))'.dependencies]
-sha2 = { version = "0.10", features = ["asm"] }
-
-[target.'cfg(any(not(any(target_arch = "x86_64", target_arch = "x86", target_arch = "aarch64")), target_env = "msvc"))'.dependencies]
-sha2 = { version = "0.10" }