feat: Add support for flow from Celestia (#4668)

### Description

Add support for flow from Celestia

### Related issues

- Fixes https://github.com/hyperlane-xyz/hyperlane-monorepo/issues/4646

### Backward compatibility

Yes

### Testing

Local run of Scraper
Check that transaction and messages can be parsed

---------

Co-authored-by: Danil Nemirovsky <4614623+ameten@users.noreply.github.com>

rust/main/chains/hyperlane-cosmos/src/error.rs

@@ -53,9 +53,24 @@ pub enum HyperlaneCosmosError {
     /// Public key error
     #[error("{0}")]
     PublicKeyError(String),
+    /// Address error
+    #[error("{0}")]
+    AddressError(String),
     /// Signer info error
     #[error("{0}")]
     SignerInfoError(String),
+    /// Serde error
+    #[error("{0}")]
+    SerdeError(#[from] serde_json::Error),
+    /// Empty error
+    #[error("{0}")]
+    UnparsableEmptyField(String),
+    /// Parsing error
+    #[error("{0}")]
+    ParsingFailed(String),
+    /// Parsing attempt failed
+    #[error("Parsing attempt failed. (Errors: {0:?})")]
+    ParsingAttemptsFailed(Vec<HyperlaneCosmosError>),
 }
 
 impl From<HyperlaneCosmosError> for ChainCommunicationError {

rust/main/chains/hyperlane-cosmos/src/libs/account.rs

@@ -63,7 +63,7 @@ impl<'a> CosmosAccountId<'a> {
 }
 
 impl TryFrom<&CosmosAccountId<'_>> for H256 {
-    type Error = ChainCommunicationError;
+    type Error = HyperlaneCosmosError;
 
     /// Builds a H256 digest from a cosmos AccountId (Bech32 encoding)
     fn try_from(account_id: &CosmosAccountId) -> Result<Self, Self::Error> {
@@ -71,7 +71,8 @@ impl TryFrom<&CosmosAccountId<'_>> for H256 {
         let h256_len = H256::len_bytes();
         let Some(start_point) = h256_len.checked_sub(bytes.len()) else {
             // input is too large to fit in a H256
-            return Err(Overflow.into());
+            let msg = "account address is too large to fit it a H256";
+            return Err(HyperlaneCosmosError::AddressError(msg.to_owned()));
         };
         let mut empty_hash = H256::default();
         let result = empty_hash.as_bytes_mut();
@@ -81,7 +82,7 @@ impl TryFrom<&CosmosAccountId<'_>> for H256 {
 }
 
 impl TryFrom<CosmosAccountId<'_>> for H256 {
-    type Error = ChainCommunicationError;
+    type Error = HyperlaneCosmosError;
 
     /// Builds a H256 digest from a cosmos AccountId (Bech32 encoding)
     fn try_from(account_id: CosmosAccountId) -> Result<Self, Self::Error> {

rust/main/chains/hyperlane-cosmos/src/libs/address.rs

@@ -91,7 +91,8 @@ impl TryFrom<&CosmosAddress> for H256 {
     type Error = ChainCommunicationError;
 
     fn try_from(cosmos_address: &CosmosAddress) -> Result<Self, Self::Error> {
-        CosmosAccountId::new(&cosmos_address.account_id).try_into()
+        H256::try_from(CosmosAccountId::new(&cosmos_address.account_id))
+            .map_err(Into::<ChainCommunicationError>::into)
     }
 }
 

rust/main/chains/hyperlane-cosmos/src/providers/cosmos/provider.rs

@@ -3,10 +3,12 @@ use std::str::FromStr;
 use async_trait::async_trait;
 use cosmrs::cosmwasm::MsgExecuteContract;
 use cosmrs::crypto::PublicKey;
+use cosmrs::proto::traits::Message;
 use cosmrs::tx::{MessageExt, SequenceNumber, SignerInfo, SignerPublicKey};
 use cosmrs::{proto, AccountId, Any, Coin, Tx};
-use itertools::Itertools;
+use itertools::{any, cloned, Itertools};
 use once_cell::sync::Lazy;
+use serde::{Deserialize, Serialize};
 use tendermint::hash::Algorithm;
 use tendermint::Hash;
 use tendermint_rpc::{client::CompatMode, Client, HttpClient};
@@ -21,11 +23,14 @@ use hyperlane_core::{
 };
 
 use crate::grpc::{WasmGrpcProvider, WasmProvider};
+use crate::providers::cosmos::provider::parse::PacketData;
 use crate::providers::rpc::CosmosRpcClient;
 use crate::{
     ConnectionConf, CosmosAccountId, CosmosAddress, CosmosAmount, HyperlaneCosmosError, Signer,
 };
 
+mod parse;
+
 /// Exponent value for atto units (10^-18).
 const ATTO_EXPONENT: u32 = 18;
 
@@ -197,8 +202,29 @@ impl CosmosProvider {
     }
 
     /// Extract contract address from transaction.
-    /// Assumes that there is only one `MsgExecuteContract` message in the transaction
     fn contract(tx: &Tx, tx_hash: &H256) -> ChainResult<H256> {
+        // We merge two error messages together so that both of them are reported
+        match Self::contract_address_from_msg_execute_contract(tx, tx_hash) {
+            Ok(contract) => Ok(contract),
+            Err(msg_execute_contract_error) => {
+                match Self::contract_address_from_msg_recv_packet(tx, tx_hash) {
+                    Ok(contract) => Ok(contract),
+                    Err(msg_recv_packet_error) => {
+                        let errors = vec![msg_execute_contract_error, msg_recv_packet_error];
+                        let error = HyperlaneCosmosError::ParsingAttemptsFailed(errors);
+                        warn!(?tx_hash, ?error);
+                        Err(ChainCommunicationError::from_other(error))?
+                    }
+                }
+            }
+        }
+    }
+
+    /// Assumes that there is only one `MsgExecuteContract` message in the transaction
+    fn contract_address_from_msg_execute_contract(
+        tx: &Tx,
+        tx_hash: &H256,
+    ) -> Result<H256, HyperlaneCosmosError> {
         use cosmrs::proto::cosmwasm::wasm::v1::MsgExecuteContract as ProtoMsgExecuteContract;
 
         let contract_execution_messages = tx
@@ -211,23 +237,45 @@ impl CosmosProvider {
 
         let contract_execution_messages_len = contract_execution_messages.len();
         if contract_execution_messages_len > 1 {
-            let msg = "transaction contains multiple contract execution messages, we are indexing the first entry only";
-            warn!(?tx_hash, ?contract_execution_messages, msg);
-            Err(ChainCommunicationError::CustomError(msg.to_owned()))?
+            let msg = "transaction contains multiple contract execution messages";
+            Err(HyperlaneCosmosError::ParsingFailed(msg.to_owned()))?
         }
 
         let any = contract_execution_messages.first().ok_or_else(|| {
             let msg = "could not find contract execution message";
-            warn!(?tx_hash, msg);
-            ChainCommunicationError::from_other_str(msg)
+            HyperlaneCosmosError::ParsingFailed(msg.to_owned())
         })?;
         let proto =
             ProtoMsgExecuteContract::from_any(any).map_err(Into::<HyperlaneCosmosError>::into)?;
         let msg = MsgExecuteContract::try_from(proto)?;
         let contract = H256::try_from(CosmosAccountId::new(&msg.contract))?;
+
         Ok(contract)
     }
 
+    fn contract_address_from_msg_recv_packet(
+        tx: &Tx,
+        tx_hash: &H256,
+    ) -> Result<H256, HyperlaneCosmosError> {
+        let packet_data = tx
+            .body
+            .messages
+            .iter()
+            .filter(|a| a.type_url == "/ibc.core.channel.v1.MsgRecvPacket")
+            .map(PacketData::try_from)
+            .flat_map(|r| r.ok())
+            .next()
+            .ok_or_else(|| {
+                let msg = "could not find IBC receive packets message containing receiver address";
+                HyperlaneCosmosError::ParsingFailed(msg.to_owned())
+            })?;
+
+        let account_id = AccountId::from_str(&packet_data.receiver)?;
+        let address = H256::try_from(CosmosAccountId::new(&account_id))?;
+
+        Ok(address)
+    }
+
     /// Reports if transaction contains fees expressed in unsupported denominations
     /// The only denomination we support at the moment is the one we express gas minimum price
     /// in the configuration of a chain. If fees contain an entry in a different denomination,

rust/main/chains/hyperlane-cosmos/src/providers/cosmos/provider/parse.rs

@@ -0,0 +1,163 @@
+use cosmrs::proto::ibc::core::channel::v1::MsgRecvPacket;
+use cosmrs::proto::prost::Message;
+use cosmrs::Any;
+use serde::{Deserialize, Serialize};
+
+use crate::HyperlaneCosmosError;
+
+#[derive(Debug, Serialize, Deserialize, Default)]
+pub struct PacketData {
+    pub amount: String,
+    pub denom: String,
+    pub memo: String,
+    pub receiver: String,
+    pub sender: String,
+}
+
+impl TryFrom<&Any> for PacketData {
+    type Error = HyperlaneCosmosError;
+
+    fn try_from(any: &Any) -> Result<Self, Self::Error> {
+        let vec = any.value.as_slice();
+        let msg = MsgRecvPacket::decode(vec).map_err(Into::<HyperlaneCosmosError>::into)?;
+        let packet = msg
+            .packet
+            .ok_or(HyperlaneCosmosError::UnparsableEmptyField(
+                "MsgRecvPacket packet is empty".to_owned(),
+            ))?;
+        let data = serde_json::from_slice::<PacketData>(&packet.data)?;
+        Ok(data)
+    }
+}
+
+impl TryFrom<Any> for PacketData {
+    type Error = HyperlaneCosmosError;
+
+    fn try_from(any: Any) -> Result<Self, Self::Error> {
+        Self::try_from(&any)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use cosmrs::proto::ibc::core::channel::v1::MsgRecvPacket;
+    use cosmrs::proto::ibc::core::channel::v1::Packet;
+    use cosmrs::proto::prost::Message;
+    use cosmrs::Any;
+
+    use crate::providers::cosmos::provider::parse::PacketData;
+    use crate::HyperlaneCosmosError;
+
+    #[test]
+    fn success() {
+        // given
+        let json = r#"{"amount":"59743800","denom":"utia","memo":"{\"wasm\":{\"contract\":\"neutron1jyyjd3x0jhgswgm6nnctxvzla8ypx50tew3ayxxwkrjfxhvje6kqzvzudq\",\"msg\":{\"transfer_remote\":{\"dest_domain\":42161,\"recipient\":\"0000000000000000000000008784aca75a95696fec93184b1c7b2d3bf5838df9\",\"amount\":\"59473800\"}},\"funds\":[{\"amount\":\"59743800\",\"denom\":\"ibc/773B4D0A3CD667B2275D5A4A7A2F0909C0BA0F4059C0B9181E680DDF4965DCC7\"}]}}","receiver":"neutron1jyyjd3x0jhgswgm6nnctxvzla8ypx50tew3ayxxwkrjfxhvje6kqzvzudq","sender":"celestia19ns7dd07g5vvrueyqlkvn4dmxt957zcdzemvj6"}"#;
+        let any = any(json);
+
+        // when
+        let data = PacketData::try_from(&any);
+
+        // then
+        assert!(data.is_ok());
+    }
+
+    #[test]
+    fn fail_json() {
+        // given
+        let json = r#"{"amount":"27000000","denom":"utia","receiver":"neutron13uuq6vgenxan43ngscjlew8lc2z32znx9qfk0n","sender":"celestia1rh4gplea4gzvaaejew8jfvp9r0qkdmfgkf55qy"}"#;
+        let any = any(json);
+
+        // when
+        let data = PacketData::try_from(&any);
+
+        // then
+        assert!(data.is_err());
+        assert!(matches!(
+            data.err().unwrap(),
+            HyperlaneCosmosError::SerdeError(_),
+        ));
+    }
+
+    #[test]
+    fn fail_empty() {
+        // given
+        let any = empty();
+
+        // when
+        let data = PacketData::try_from(&any);
+
+        // then
+        assert!(data.is_err());
+        assert!(matches!(
+            data.err().unwrap(),
+            HyperlaneCosmosError::UnparsableEmptyField(_),
+        ));
+    }
+
+    #[test]
+    fn fail_decode() {
+        // given
+        let any = wrong_encoding();
+
+        // when
+        let data = PacketData::try_from(&any);
+
+        // then
+        assert!(data.is_err());
+        assert!(matches!(
+            data.err().unwrap(),
+            HyperlaneCosmosError::Prost(_),
+        ));
+    }
+
+    fn any(json: &str) -> Any {
+        let packet = Packet {
+            sequence: 0,
+            source_port: "".to_string(),
+            source_channel: "".to_string(),
+            destination_port: "".to_string(),
+            destination_channel: "".to_string(),
+            data: json.as_bytes().to_vec(),
+            timeout_height: None,
+            timeout_timestamp: 0,
+        };
+
+        let msg = MsgRecvPacket {
+            packet: Option::from(packet),
+            proof_commitment: vec![],
+            proof_height: None,
+            signer: "".to_string(),
+        };
+
+        encode_proto(&msg)
+    }
+
+    fn empty() -> Any {
+        let msg = MsgRecvPacket {
+            packet: None,
+            proof_commitment: vec![],
+            proof_height: None,
+            signer: "".to_string(),
+        };
+
+        encode_proto(&msg)
+    }
+
+    fn wrong_encoding() -> Any {
+        let buf = vec![1, 2, 3];
+        Any {
+            type_url: "".to_string(),
+            value: buf,
+        }
+    }
+
+    fn encode_proto(msg: &MsgRecvPacket) -> Any {
+        let mut buf = Vec::with_capacity(msg.encoded_len());
+        MsgRecvPacket::encode(&msg, &mut buf).unwrap();
+
+        Any {
+            type_url: "".to_string(),
+            value: buf,
+        }
+    }
+}