Setup binary for testing full flow

packages/ciphernode/core/src/ciphernode.rs

@@ -3,6 +3,7 @@ use crate::{
     eventbus::EventBus,
     events::{ComputationRequested, EnclaveEvent, KeyshareCreated},
     fhe::{Fhe, GenerateKeyshare},
+    Subscribe,
 };
 use actix::prelude::*;
 use anyhow::Result;
@@ -21,6 +22,12 @@ impl Ciphernode {
     pub fn new(bus: Addr<EventBus>, fhe: Addr<Fhe>, data: Addr<Data>) -> Self {
         Self { bus, fhe, data }
     }
+
+    pub fn attach(bus: Addr<EventBus>, fhe: Addr<Fhe>, data: Addr<Data>) -> Addr<Self> {
+        let node = Ciphernode::new(bus.clone(), fhe, data).start();
+        bus.do_send(Subscribe::new("ComputationRequested", node.clone().into()));
+        node
+    }
 }
 
 impl Handler<EnclaveEvent> for Ciphernode {
@@ -74,7 +81,6 @@ async fn on_computation_requested(
 
     // broadcast the KeyshareCreated message
     let event = EnclaveEvent::from(KeyshareCreated { pubkey, e3_id });
-
     bus.do_send(event);
 
     Ok(())

packages/ciphernode/core/src/committee.rs

@@ -7,6 +7,7 @@ use crate::{
     eventbus::EventBus,
     events::{E3id, EnclaveEvent},
     fhe::Fhe,
+    Subscribe,
 };
 
 pub struct CommitteeManager {
@@ -27,6 +28,16 @@ impl CommitteeManager {
             keys: HashMap::new(),
         }
     }
+
+    pub fn attach(bus: Addr<EventBus>, fhe: Addr<Fhe>) -> Addr<Self> {
+        let addr = CommitteeManager::new(bus.clone(), fhe).start();
+        bus.do_send(Subscribe::new(
+            "ComputationRequested",
+            addr.clone().recipient(),
+        ));
+        bus.do_send(Subscribe::new("KeyshareCreated", addr.clone().into()));
+        addr
+    }
 }
 
 impl Handler<EnclaveEvent> for CommitteeManager {
@@ -50,16 +61,15 @@ impl Handler<EnclaveEvent> for CommitteeManager {
                 if let Some(key) = self.keys.get(&data.e3_id) {
                     key.do_send(data);
                 }
-            },
+            }
             EnclaveEvent::PublicKeyAggregated { data, .. } => {
                 let Some(key) = self.keys.get(&data.e3_id) else {
                     return;
                 };
 
                 key.do_send(Die);
                 self.keys.remove(&data.e3_id);
-            }
-            // _ => (),
+            } // _ => (),
         }
     }
 }

packages/ciphernode/core/src/eventbus.rs

@@ -83,7 +83,7 @@ impl Handler<EnclaveEvent> for EventBus {
             // We have seen this before
             return;
         }
-
+        
         // TODO: How can we ensure the event we see is coming in in the correct order?
         if let Some(listeners) = self.listeners.get("*") {
             for listener in listeners {

packages/ciphernode/core/src/events.rs

@@ -1,3 +1,4 @@
+use crate::fhe::{WrappedPublicKey, WrappedPublicKeyShare};
 use actix::Message;
 use bincode;
 use serde::{Deserialize, Serialize};
@@ -7,8 +8,6 @@ use std::{
     hash::{DefaultHasher, Hash, Hasher},
 };
 
-use crate::fhe::{WrappedPublicKey, WrappedPublicKeyShare};
-
 #[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub struct E3id(pub String);
 impl fmt::Display for E3id {
@@ -23,6 +22,12 @@ impl E3id {
     }
 }
 
+impl From<u32> for E3id {
+    fn from(value: u32) -> Self {
+        E3id::new(value.to_string())
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub struct EventId(pub [u8; 32]);
 
@@ -75,7 +80,7 @@ impl EnclaveEvent {
     }
 
     pub fn get_id(&self) -> EventId {
-       self.clone().into() 
+        self.clone().into()
     }
 }
 
@@ -116,6 +121,12 @@ impl From<PublicKeyAggregated> for EnclaveEvent {
     }
 }
 
+impl fmt::Display for EnclaveEvent {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(&format!("{}({})", self.event_type(), self.get_id()))
+    }
+}
+
 #[derive(Message, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
 #[rtype(result = "anyhow::Result<()>")]
 pub struct KeyshareCreated {
@@ -163,7 +174,18 @@ impl EnclaveEvent {
 #[cfg(test)]
 mod tests {
 
-    use crate::events::extract_enclave_event_name;
+    use std::error::Error;
+
+    use fhe::{
+        bfv::{BfvParametersBuilder, SecretKey},
+        mbfv::{CommonRandomPoly, PublicKeyShare},
+    };
+    use rand::SeedableRng;
+    use rand_chacha::ChaCha20Rng;
+
+    use crate::{events::extract_enclave_event_name, E3id, KeyshareCreated, WrappedPublicKeyShare};
+
+    use super::EnclaveEvent;
 
     #[test]
     fn test_extract_enum_name() {
@@ -176,4 +198,29 @@ mod tests {
             "CommitteeSelected"
         );
     }
+
+    #[test]
+    fn test_deserialization() -> Result<(), Box<dyn Error>> {
+        let moduli = &vec![0x3FFFFFFF000001];
+        let degree = 2048usize;
+        let plaintext_modulus = 1032193u64;
+        let mut rng = ChaCha20Rng::from_entropy();
+        let params = BfvParametersBuilder::new()
+            .set_degree(degree)
+            .set_plaintext_modulus(plaintext_modulus)
+            .set_moduli(&moduli)
+            .build_arc()?;
+        let crp = CommonRandomPoly::new(&params, &mut rng)?;
+        let sk_share = { SecretKey::random(&params, &mut rng) };
+        let pk_share = { PublicKeyShare::new(&sk_share, crp.clone(), &mut rng)? };
+        let pubkey = WrappedPublicKeyShare::from_fhe_rs(pk_share, params.clone(), crp.clone());
+        let kse = EnclaveEvent::from(KeyshareCreated {
+            e3_id: E3id::from(1001),
+            pubkey,
+        });
+        let kse_bytes = kse.to_bytes()?;
+        let _ = EnclaveEvent::from_bytes(&kse_bytes.clone());
+        // deserialization occurred without panic!
+        Ok(())
+    }
 }

packages/ciphernode/core/src/fhe.rs

@@ -113,7 +113,7 @@ impl serde::Serialize for WrappedPublicKeyShare {
         use serde::ser::SerializeStruct;
         let bytes = self.inner.to_bytes();
         let par_bytes = self.params.to_bytes();
-        let crp_bytes = self.params.to_bytes();
+        let crp_bytes = self.crp.to_bytes();
         // Intermediate struct of bytes
         let mut state = serializer.serialize_struct("PublicKeyShare", 2)?;
         state.serialize_field("par_bytes", &par_bytes)?;
@@ -123,6 +123,7 @@ impl serde::Serialize for WrappedPublicKeyShare {
     }
 }
 
+
 /// Wrapped PublicKey. This is wrapped to provide an inflection point
 /// as we use this library elsewhere we only implement traits as we need them
 /// and avoid exposing underlying structures from fhe.rs

packages/ciphernode/core/src/lib.rs

@@ -12,6 +12,19 @@ mod events;
 mod fhe;
 mod ordered_set;
 mod p2p;
+mod logger;
+
+// TODO: this is too permissive
+pub use data::*;
+pub use ciphernode::*;
+pub use committee::*;
+pub use committee_key::*;
+pub use eventbus::*;
+pub use events::*;
+pub use fhe::*;
+pub use p2p::*;
+pub use actix::prelude::*;
+pub use logger::*;
 
 pub use data::*;
 pub use ciphernode::*;

packages/ciphernode/core/src/logger.rs

@@ -0,0 +1,27 @@
+use actix::{Actor, Addr, Context, Handler};
+
+use crate::{EnclaveEvent, EventBus, Subscribe};
+
+pub struct SimpleLogger;
+
+impl SimpleLogger {
+    pub fn attach(bus:Addr<EventBus>) -> Addr<Self>{
+       let addr = Self.start();
+       bus.do_send(Subscribe { 
+           listener:addr.clone().recipient(),
+           event_type: "*".to_string() 
+       });
+       addr
+    }
+}
+
+impl Actor for SimpleLogger {
+    type Context = Context<Self>;
+}
+
+impl Handler<EnclaveEvent> for SimpleLogger {
+    type Result = ();
+    fn handle(&mut self, msg: EnclaveEvent, _: &mut Self::Context) -> Self::Result {
+        println!("{}", msg);
+    }
+}

packages/ciphernode/core/src/p2p.rs

@@ -2,12 +2,7 @@ use std::{collections::HashSet, error::Error};
 
 /// Actor for connecting to an libp2p client via it's mpsc channel interface
 /// This Actor should be responsible for
-/// 1. Sending and Recieving Vec<u8> messages with libp2p
-/// 2. Converting between Vec<u8> and EnclaveEvents::Xxxxxxxxx()
-/// 3. Broadcasting over the local eventbus
-/// 4. Listening to the local eventbus for messages to be published to libp2p
 use actix::prelude::*;
-use anyhow::anyhow;
 use p2p::EnclaveRouter;
 use tokio::sync::mpsc::{Receiver, Sender};
 

packages/ciphernode/enclave_node/src/bin/ciphernode.rs

@@ -1,18 +1,21 @@
-use std::error::Error;
-
 use enclave_core::Actor;
 use enclave_core::Ciphernode;
+use enclave_core::CommitteeManager;
 use enclave_core::Data;
 use enclave_core::EventBus;
 use enclave_core::Fhe;
 use enclave_core::P2p;
+use enclave_core::SimpleLogger;
+use std::error::Error;
 
 #[actix_rt::main]
 async fn main() -> Result<(), Box<dyn Error>> {
     let fhe = Fhe::try_default()?.start();
     let bus = EventBus::new(true).start();
     let data = Data::new(true).start(); // TODO: Use a sled backed Data Actor
-    let _node = Ciphernode::new(bus.clone(), fhe.clone(), data.clone()).start();
+    SimpleLogger::attach(bus.clone());
+    Ciphernode::attach(bus.clone(), fhe.clone(), data.clone());
+    CommitteeManager::attach(bus.clone(), fhe.clone());
     let (_, h) = P2p::spawn_libp2p(bus.clone())?;
     println!("Ciphernode");
     let _ = tokio::join!(h);

packages/ciphernode/enclave_node/src/bin/cmd.rs

@@ -0,0 +1,39 @@
+use std::error::Error;
+
+use enclave_core::Actor;
+use enclave_core::ComputationRequested;
+use enclave_core::E3id;
+use enclave_core::EnclaveEvent;
+use enclave_core::EventBus;
+use enclave_core::P2p;
+use tokio::{
+    self,
+    io::{self, AsyncBufReadExt, BufReader},
+};
+
+#[actix_rt::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let bus = EventBus::new(true).start();
+    let (_, t1) = P2p::spawn_libp2p(bus.clone())?;
+    let mut stdin = BufReader::new(io::stdin()).lines();
+    let t2 = tokio::spawn(async move {
+        let mut id: u32 = 1000;
+        while let Ok(Some(line)) = stdin.next_line().await {
+            match line.as_str() {
+                "test" => {
+                    id += 1;
+                    bus.do_send(EnclaveEvent::from(ComputationRequested {
+                        e3_id: E3id::from(id),
+                        nodecount: 3,
+                        threshold: 3,
+                        sortition_seed: 100,
+                    }));
+                }
+                _ => println!("Unknown command"),
+            }
+        }
+    });
+
+    let _ = tokio::join!(t1, t2);
+    Ok(())
+}

packages/ciphernode/p2p/src/lib.rs

@@ -127,7 +127,6 @@ impl EnclaveRouter {
         loop {
             select! {
                 Some(line) = self.cmd_rx.recv() => {
-                    println!("Receiving input");
                     if let Err(e) = self.swarm.as_mut().unwrap()
                         .behaviour_mut().gossipsub
                         .publish(self.topic.as_mut().unwrap().clone(), line) {
@@ -153,9 +152,9 @@ impl EnclaveRouter {
                         message,
                     })) => {
                         println!(
-                            "Got message: '{}' with id: {id} from peer: {peer_id}",
-                            String::from_utf8_lossy(&message.data),
+                            "Got message with id: {id} from peer: {peer_id}",
                         );
+                        println!("{:?}", message);
                         self.evt_tx.send(message.data).await?;
                     },
                     SwarmEvent::NewListenAddr { address, .. } => {