Merge remote-tracking branch 'origin/upgrade-error-stack' into upgrad…

…e-error-stack

ampd/src/config.rs

@@ -2,6 +2,7 @@ use serde::Deserialize;
 
 use crate::broadcaster;
 use crate::evm::{deserialize_evm_chain_configs, EvmChainConfig};
+use crate::handlers::multisig::MultisigConfig;
 use crate::tofnd::Config as TofndConfig;
 use crate::url::Url;
 
@@ -15,6 +16,7 @@ pub struct Config {
     pub evm_chains: Vec<EvmChainConfig>,
     pub tofnd_config: TofndConfig,
     pub event_buffer_cap: usize,
+    pub multisig: Option<MultisigConfig>,
 }
 
 impl Default for Config {
@@ -26,6 +28,7 @@ impl Default for Config {
             evm_chains: vec![],
             tofnd_config: TofndConfig::default(),
             event_buffer_cap: 100000,
+            multisig: None,
         }
     }
 }

ampd/src/handlers/errors.rs

@@ -8,4 +8,6 @@ pub enum Error {
     Finalizer,
     #[error("failed to deserialize the event")]
     DeserializeEvent,
+    #[error("failed to get signature from tofnd")]
+    Sign,
 }

ampd/src/handlers/multisig.rs

@@ -1,26 +1,40 @@
 use std::collections::HashMap;
+use std::convert::TryInto;
 
+use async_trait::async_trait;
+use cosmrs::cosmwasm::MsgExecuteContract;
+use cosmwasm_std::{HexBinary, Uint64};
 use ecdsa::VerifyingKey;
-use hex::FromHex;
+use error_stack::{IntoReport, ResultExt};
+use hex::{encode, FromHex};
 use serde::de::Error as DeserializeError;
 use serde::{Deserialize, Deserializer};
+use tracing::info;
 
+use events::Error::EventTypeMismatch;
 use events_derive;
-use multisig::types::KeyID;
+use events_derive::try_from;
+use multisig::msg::ExecuteMsg;
 
+use crate::event_processor::EventHandler;
+use crate::handlers::errors::Error::{self, DeserializeEvent};
+use crate::queue::queued_broadcaster::BroadcasterClient;
+use crate::tofnd::grpc::SharableEcdsaClient;
 use crate::tofnd::MessageDigest;
 use crate::types::PublicKey;
 use crate::types::TMAddress;
-use events_derive::try_from;
 
-#[allow(dead_code)]
+#[derive(Debug, Deserialize)]
+pub struct MultisigConfig {
+    pub address: TMAddress,
+}
+
 #[derive(Debug, Deserialize)]
 #[try_from("wasm-signing_started")]
 struct SigningStartedEvent {
     #[serde(rename = "_contract_address")]
     contract_address: TMAddress,
     session_id: u64,
-    key_id: KeyID,
     #[serde(deserialize_with = "deserialize_public_keys")]
     pub_keys: HashMap<TMAddress, PublicKey>,
     #[serde(with = "hex")]
@@ -52,30 +66,144 @@ where
         .collect()
 }
 
+pub struct Handler<B>
+where
+    B: BroadcasterClient,
+{
+    worker: TMAddress,
+    multisig: TMAddress,
+    broadcaster: B,
+    signer: SharableEcdsaClient,
+}
+
+impl<B> Handler<B>
+where
+    B: BroadcasterClient,
+{
+    pub fn new(
+        worker: TMAddress,
+        multisig: TMAddress,
+        broadcaster: B,
+        signer: SharableEcdsaClient,
+    ) -> Self {
+        Self {
+            worker,
+            multisig,
+            broadcaster,
+            signer,
+        }
+    }
+
+    async fn broadcast_signature(
+        &self,
+        session_id: impl Into<Uint64>,
+        signature: impl Into<HexBinary>,
+    ) -> error_stack::Result<(), Error> {
+        let msg = serde_json::to_vec(&ExecuteMsg::SubmitSignature {
+            session_id: session_id.into(),
+            signature: signature.into(),
+        })
+        .expect("submit signature msg should serialize");
+
+        let tx = MsgExecuteContract {
+            sender: self.worker.as_ref().clone(),
+            contract: self.multisig.as_ref().clone(),
+            msg,
+            funds: vec![],
+        };
+
+        self.broadcaster
+            .broadcast(tx)
+            .await
+            .change_context(Error::Broadcaster)
+    }
+}
+
+#[async_trait]
+impl<B> EventHandler for Handler<B>
+where
+    B: BroadcasterClient + Send + Sync,
+{
+    type Err = Error;
+
+    async fn handle(&self, event: &events::Event) -> error_stack::Result<(), Error> {
+        let SigningStartedEvent {
+            contract_address,
+            session_id,
+            pub_keys,
+            msg,
+        } = match event.try_into() as error_stack::Result<_, _> {
+            Err(report) if matches!(report.current_context(), EventTypeMismatch(_)) => {
+                return Ok(());
+            }
+            result => result.change_context(DeserializeEvent)?,
+        };
+
+        if self.multisig != contract_address {
+            return Ok(());
+        }
+
+        info!(
+            session_id = session_id,
+            msg = encode(&msg),
+            "get signing request",
+        );
+
+        match pub_keys.get(&self.worker) {
+            Some(pub_key) => {
+                let signature = self
+                    .signer
+                    .sign(self.multisig.to_string().as_str(), msg, pub_key)
+                    .await
+                    .change_context(Error::Sign)?;
+
+                info!(signature = encode(&signature), "ready to submit signature");
+
+                self.broadcast_signature(session_id, signature).await?;
+
+                Ok(())
+            }
+            None => {
+                info!("worker is not a participant");
+                Ok(())
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod test {
-    use error_stack::Result;
-    use std::collections::HashMap;
-    use std::convert::{TryFrom, TryInto};
 
     use base64::engine::general_purpose::STANDARD;
     use base64::Engine;
+    use std::collections::HashMap;
+    use std::convert::{TryFrom, TryInto};
+    use std::time::Duration;
+
+    use cosmos_sdk_proto::cosmos::base::abci::v1beta1::TxResponse;
+    use cosmrs::{AccountId, Gas};
     use cosmwasm_std::{Addr, HexBinary, Uint64};
     use ecdsa::SigningKey;
+    use error_stack::{Report, Result};
     use rand::rngs::OsRng;
     use tendermint::abci;
 
-    use multisig::events::Event::SigningStarted;
-    use multisig::types::{MsgToSign, PublicKey};
-
     use super::*;
+    use crate::broadcaster::MockBroadcaster;
+    use crate::queue::queued_broadcaster::{QueuedBroadcaster, QueuedBroadcasterClient};
+    use crate::tofnd;
+    use crate::tofnd::grpc::{MockEcdsaClient, SharableEcdsaClient};
     use crate::types;
+    use multisig::events::Event::SigningStarted;
+    use multisig::types::{KeyID, MsgToSign, PublicKey};
+
+    const MULTISIG_ADDRESS: &str = "axelarvaloper1zh9wrak6ke4n6fclj5e8yk397czv430ygs5jz7";
 
-    fn rand_account() -> String {
+    fn rand_account() -> TMAddress {
         types::PublicKey::from(SigningKey::random(&mut OsRng).verifying_key())
             .account_id("axelar")
             .unwrap()
-            .to_string()
+            .into()
     }
 
     fn rand_public_key() -> PublicKey {
@@ -91,7 +219,7 @@ mod test {
 
     fn signing_started_event() -> events::Event {
         let pub_keys = (0..10)
-            .map(|_| (rand_account(), rand_public_key()))
+            .map(|_| (rand_account().to_string(), rand_public_key()))
             .collect::<HashMap<String, PublicKey>>();
 
         let poll_started = SigningStarted {
@@ -106,10 +234,7 @@ mod test {
 
         let mut event: cosmwasm_std::Event = poll_started.into();
         event.ty = format!("wasm-{}", event.ty);
-        event = event.add_attribute(
-            "_contract_address",
-            "axelarvaloper1zh9wrak6ke4n6fclj5e8yk397czv430ygs5jz7",
-        );
+        event = event.add_attribute("_contract_address", MULTISIG_ADDRESS);
 
         events::Event::try_from(abci::Event::new(
             event.ty,
@@ -123,6 +248,22 @@ mod test {
         .unwrap()
     }
 
+    fn get_handler(
+        worker: TMAddress,
+        multisig: TMAddress,
+        signer: SharableEcdsaClient,
+    ) -> Handler<QueuedBroadcasterClient> {
+        let mut broadcaster = MockBroadcaster::new();
+        broadcaster
+            .expect_broadcast()
+            .returning(|_| Ok(TxResponse::default()));
+
+        let (broadcaster, _) =
+            QueuedBroadcaster::new(broadcaster, Gas::default(), 100, Duration::from_secs(5));
+
+        Handler::new(worker, multisig, broadcaster.client(), signer)
+    }
+
     #[test]
     fn should_not_deserialize_incorrect_event_type() {
         // incorrect event type
@@ -150,7 +291,7 @@ mod test {
         let invalid_pub_key: [u8; 32] = rand::random();
         let mut map: HashMap<String, PublicKey> = HashMap::new();
         map.insert(
-            rand_account(),
+            rand_account().to_string(),
             PublicKey::unchecked(HexBinary::from(invalid_pub_key.as_slice())),
         );
         match event {
@@ -177,4 +318,58 @@ mod test {
 
         assert!(event.is_ok());
     }
+
+    #[tokio::test]
+    async fn should_not_handle_event_if_multisig_address_does_not_match() {
+        let mut client = MockEcdsaClient::new();
+        client
+            .expect_sign()
+            .returning(move |_, _, _| Err(Report::from(tofnd::error::Error::SignFailed)));
+
+        let handler = get_handler(
+            rand_account(),
+            rand_account(),
+            SharableEcdsaClient::new(client),
+        );
+
+        assert!(handler.handle(&signing_started_event()).await.is_ok());
+    }
+
+    #[tokio::test]
+    async fn should_not_handle_event_if_worker_is_not_a_participant() {
+        let mut client = MockEcdsaClient::new();
+        client
+            .expect_sign()
+            .returning(move |_, _, _| Err(Report::from(tofnd::error::Error::SignFailed)));
+
+        let handler = get_handler(
+            rand_account(),
+            TMAddress::from(MULTISIG_ADDRESS.parse::<AccountId>().unwrap()),
+            SharableEcdsaClient::new(client),
+        );
+
+        assert!(handler.handle(&signing_started_event()).await.is_ok());
+    }
+
+    #[tokio::test]
+    async fn should_not_handle_event_if_sign_failed() {
+        let mut client = MockEcdsaClient::new();
+        client
+            .expect_sign()
+            .returning(move |_, _, _| Err(Report::from(tofnd::error::Error::SignFailed)));
+
+        let event = signing_started_event();
+        let signing_started: SigningStartedEvent = ((&event).try_into() as Result<_, _>).unwrap();
+        let worker = signing_started.pub_keys.keys().next().unwrap().clone();
+        let handler = get_handler(
+            worker,
+            TMAddress::from(MULTISIG_ADDRESS.parse::<AccountId>().unwrap()),
+            SharableEcdsaClient::new(client),
+        );
+
+        assert!(matches!(
+            *handler.handle(&event).await.unwrap_err().current_context(),
+            Error::Sign
+        ));
+    }
 }