WIP

common/crypto/dkg/src/bte/encryption.rs

@@ -14,8 +14,8 @@ use std::collections::HashMap;
 use std::ops::Neg;
 use zeroize::Zeroize;
 
-#[derive(Debug)]
-#[cfg_attr(test, derive(Clone, PartialEq))]
+#[derive(Debug, Clone)]
+#[cfg_attr(test, derive(PartialEq))]
 pub struct Ciphertexts {
     pub rr: [G1Projective; NUM_CHUNKS],
     pub ss: [G1Projective; NUM_CHUNKS],

common/crypto/dkg/src/bte/proof_chunking.rs

@@ -67,8 +67,8 @@ impl<'a> Instance<'a> {
     }
 }
 
-#[derive(Debug)]
-#[cfg_attr(test, derive(Clone, PartialEq))]
+#[derive(Debug, Clone)]
+#[cfg_attr(test, derive(PartialEq))]
 pub struct ProofOfChunking {
     y0: G1Projective,
     bb: Vec<G1Projective>,

common/crypto/dkg/src/bte/proof_sharing.rs

@@ -76,8 +76,8 @@ impl<'a> Instance<'a> {
     }
 }
 
-#[derive(Debug)]
-#[cfg_attr(test, derive(Clone, PartialEq))]
+#[derive(Debug, Clone)]
+#[cfg_attr(test, derive(PartialEq))]
 pub struct ProofOfSecretSharing {
     ff: G1Projective,
     aa: G2Projective,

common/crypto/dkg/src/dealing.rs

@@ -17,7 +17,7 @@ use rand_core::RngCore;
 use std::collections::BTreeMap;
 use zeroize::Zeroize;
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 #[cfg_attr(test, derive(PartialEq))]
 pub struct Dealing {
     pub public_coefficients: PublicCoefficients,

validator-api/src/dkg/networking/codec.rs

@@ -9,6 +9,7 @@ use tokio_util::codec::{Decoder, Encoder};
 
 const MAX_ALLOWED_MESSAGE_LEN: usize = 2 * 1024 * 1024;
 
+#[derive(Debug)]
 pub struct DkgCodec;
 
 impl Encoder<OffchainDkgMessage> for DkgCodec {

validator-api/src/dkg/networking/handler.rs

@@ -0,0 +1,104 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::dkg::networking::codec::DkgCodec;
+use crate::dkg::networking::message::{ErrorReason, OffchainDkgMessage};
+use crate::dkg::state::DkgState;
+use futures::StreamExt;
+use std::net::SocketAddr;
+use std::time::Duration;
+use tokio::net::TcpStream;
+use tokio::time::timeout;
+use tokio_util::codec::Framed;
+
+const DEFAULT_MAX_CONNECTION_DURATION: Duration = Duration::new(2 * 60, 0);
+
+#[derive(Debug)]
+pub(crate) struct ConnectionHandler {
+    // connection cannot exist for more than this time
+    max_connection_duration: Duration,
+    dkg_state: DkgState,
+    conn: Framed<TcpStream, DkgCodec>,
+    remote: SocketAddr,
+}
+
+impl ConnectionHandler {
+    pub(crate) fn new(dkg_state: DkgState, conn: TcpStream, remote: SocketAddr) -> Self {
+        ConnectionHandler {
+            max_connection_duration: DEFAULT_MAX_CONNECTION_DURATION,
+            dkg_state,
+            remote,
+            conn: Framed::new(conn, DkgCodec),
+        }
+    }
+
+    async fn send_error_response<S: Into<String>>(
+        &mut self,
+        error: ErrorReason,
+        additional_info: Option<S>,
+    ) {
+        //
+    }
+
+    async fn _handle_connection(&mut self) {
+        debug!("Starting connection handler for {}", self.remote);
+
+        if !self.dkg_state.is_dealers_remote_address(self.remote).await {
+            let msg = format!(
+                "{} is not a socket address of any known dealer. Closing the connection.",
+                self.remote
+            );
+            warn!("{}", msg);
+            self.send_error_response(ErrorReason::UnknownDealer, Some(msg))
+                .await;
+            return;
+        }
+
+        while let Some(framed_dkg_request) = self.conn.next().await {
+            match framed_dkg_request {
+                Ok(framed_dkg_request) => {
+                    todo!("handle packet")
+                }
+                Err(err) => {
+                    warn!(
+                        "The socket connection got corrupted with error: {:?}. Closing the socket",
+                        err
+                    );
+                    break;
+                }
+            }
+        }
+
+        debug!("Closing connection from {}", self.remote);
+    }
+
+    pub(crate) async fn handle_connection(mut self) {
+        let remote = self.remote;
+        if timeout(self.max_connection_duration, self._handle_connection())
+            .await
+            .is_err()
+        {
+            warn!(
+                "we timed out while trying to resolve connection from {}",
+                remote
+            );
+            self.send_error_response(
+                ErrorReason::Timeout,
+                Some(format!(
+                    "could not resolve connection within {:?}",
+                    self.max_connection_duration
+                )),
+            )
+            .await;
+        }
+    }
+
+    async fn handle_request(&self, request: OffchainDkgMessage) {
+        match request {
+            OffchainDkgMessage::NewDealing { .. } => {}
+            OffchainDkgMessage::RemoteDealingRequest { .. } => {}
+            OffchainDkgMessage::RemoteDealingResponse { .. } => {}
+            OffchainDkgMessage::ErrorResponse { .. } => {}
+        }
+    }
+}

validator-api/src/dkg/networking/listener.rs

@@ -3,47 +3,29 @@
 
 // TODO: if it becomes too cumbersome, perhaps consider a more streamlined solution like tarpc
 
-use crate::dkg::networking::codec::DkgCodec;
-use futures::StreamExt;
+use crate::dkg::networking::handler::ConnectionHandler;
+use crate::dkg::state::DkgState;
 use std::fmt::Display;
 use std::net::SocketAddr;
 use std::process;
 use tokio::net::{TcpListener, TcpStream, ToSocketAddrs};
-use tokio_util::codec::Framed;
 
 // note that we do not expect persistent connections between dealers, they should only really
 // exist for the duration of a single message exchange
 pub(crate) struct Listener<A> {
     address: A,
+    dkg_state: DkgState,
 }
 
 impl<A> Listener<A> {
-    pub(crate) fn new(address: A) -> Self {
-        Listener { address }
+    pub(crate) fn new(address: A, dkg_state: DkgState) -> Self {
+        Listener { address, dkg_state }
     }
 
     fn on_connect(&self, conn: TcpStream, remote: SocketAddr) {
-        // TODO: here be a check to see if the connection originates from a known dealer
-        tokio::spawn(async move {
-            debug!("Starting connection handler for {}", remote);
-            let mut framed_conn = Framed::new(conn, DkgCodec);
-            while let Some(framed_dkg_request) = framed_conn.next().await {
-                match framed_dkg_request {
-                    Ok(framed_dkg_request) => {
-                        todo!("handle packet")
-                    }
-                    Err(err) => {
-                        warn!(
-                        "The socket connection got corrupted with error: {:?}. Closing the socket",
-                        err
-                    );
-                        break;
-                    }
-                }
-            }
-
-            debug!("Closing connection from {}", remote);
-        });
+        tokio::spawn(
+            ConnectionHandler::new(self.dkg_state.clone(), conn, remote).handle_connection(),
+        );
     }
 
     async fn run(&mut self)

validator-api/src/dkg/networking/message.rs

@@ -20,6 +20,10 @@ pub enum OffchainDkgMessage {
         id: u64,
         message: RemoteDealingResponseMessage,
     },
+    ErrorResponse {
+        id: u64,
+        message: ErrorResponseMessage,
+    },
 }
 
 pub struct NewDealingMessage {
@@ -44,6 +48,16 @@ pub enum RemoteDealingResponseMessage {
     Unavailable,
 }
 
+pub struct ErrorResponseMessage {
+    reason: ErrorReason,
+    additional_info: Option<String>,
+}
+
+pub enum ErrorReason {
+    UnknownDealer,
+    Timeout,
+}
+
 impl OffchainDkgMessage {
     fn frame(self) -> FramedOffchainDkgMessage {
         todo!()
@@ -98,7 +112,7 @@ struct FramedOffchainDkgMessage {
 
 impl FramedOffchainDkgMessage {
     fn into_bytes(mut self) -> Vec<u8> {
-        let mut header_bytes = self.header.to_bytes();
+        let mut header_bytes = self.header.into_bytes();
         let mut out = Vec::with_capacity(header_bytes.len() + self.payload.len());
 
         out.append(&mut header_bytes);
@@ -121,7 +135,7 @@ pub(crate) struct Header {
 impl Header {
     pub(crate) const LEN: usize = 13;
 
-    pub(crate) fn to_bytes(&self) -> Vec<u8> {
+    pub(crate) fn into_bytes(self) -> Vec<u8> {
         let mut out = Vec::with_capacity(Self::LEN);
         out.push(self.message_type as u8);
         out.extend_from_slice(&self.payload_length.to_be_bytes());
@@ -163,7 +177,7 @@ mod tests {
             protocol_version: PROTOCOL_VERSION,
         };
 
-        let bytes = valid_header.to_bytes();
+        let bytes = valid_header.into_bytes();
         assert_eq!(valid_header, Header::try_from_bytes(&bytes).unwrap())
     }
 }

validator-api/src/dkg/networking/mod.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 pub(crate) mod codec;
+mod handler;
 mod listener;
 pub(crate) mod message;
 

validator-api/src/dkg/state.rs

@@ -1,27 +1,68 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use coconut_dkg_common::types::NodeIndex;
+use coconut_dkg_common::types::{Epoch, NodeIndex};
 use crypto::asymmetric::identity;
-use dkg::bte;
+use dkg::{bte, Dealing};
+use futures::lock::Mutex;
 use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+type IdentityBytes = [u8; identity::PUBLIC_KEY_LENGTH];
 
 // TODO: some TryFrom impl to convert from encoded contract data
+// note: each dealer is also a receiver which simplifies some logic significantly
+#[derive(Debug)]
 struct Dealer {
-    // node_index: Option<NodeIndex>,
-    // bte_public_key: bte::PublicKey,
+    node_index: NodeIndex,
+    bte_public_key: bte::PublicKey,
     identity: identity::PublicKey,
+    remote_address: SocketAddr,
 }
 
-enum DealerState {
-    DealingReceived(),
-}
-
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Clone)]
 pub(crate) struct DkgState {
-    //
+    inner: Arc<Mutex<DkgStateInner>>,
+}
+
+#[derive(Debug)]
+struct DkgStateInner {
+    bte_decryption_key: bte::DecryptionKey,
+    signing_key: identity::PublicKey,
+
+    current_epoch: Epoch,
+
+    expected_epoch_dealing_digests: HashMap<IdentityBytes, [u8; 32]>,
+    current_epoch_dealers: HashMap<IdentityBytes, Dealer>,
+    verified_epoch_dealings: HashMap<IdentityBytes, Dealing>,
+    unconfirmed_dealings: HashMap<IdentityBytes, Dealing>,
 }
 
 impl DkgState {
     // some save/load action here
+
+    pub(crate) async fn is_dealers_remote_address(&self, remote: SocketAddr) -> bool {
+        let dealers = &self.inner.lock().await.current_epoch_dealers;
+        dealers
+            .values()
+            .any(|dealer| dealer.remote_address == remote)
+    }
+
+    pub(crate) async fn get_verified_dealing(
+        &self,
+        dealer: identity::PublicKey,
+    ) -> Option<Dealing> {
+        self.inner
+            .lock()
+            .await
+            .verified_epoch_dealings
+            .get(&dealer.to_bytes())
+            .cloned()
+    }
+}
+
+impl DkgStateInner {
+    //
 }