Connectors v1++

pallets/connectors/src/lib.rs

@@ -242,12 +242,22 @@ pub mod pallet {
 			domain: Domain,
 			router: Router<CurrencyIdOf<T>>,
 		},
+
+		IncomingMessage {
+			sender: T::AccountId,
+			message: Vec<u8>,
+		},
 	}
 
 	#[pallet::storage]
 	pub(crate) type DomainRouter<T: Config> =
 		StorageMap<_, Blake2_128Concat, Domain, Router<CurrencyIdOf<T>>>;
 
+	/// The set of known connectors. This set is used as an allow-list when authorizing
+	/// the origin of incoming messages through the `handle` extrinsic.
+	#[pallet::storage]
+	pub(crate) type KnownConnectors<T: Config> = StorageMap<_, Blake2_128Concat, T::AccountId, ()>;
+
 	#[pallet::error]
 	pub enum Error<T> {
 		/// A pool could not be found
@@ -268,6 +278,8 @@ pub mod pallet {
 		UnauthorizedTransfer,
 		/// Failed to build Ethereum_Xcm call
 		FailedToBuildEthereumXcmCall,
+		/// The origin of an incoming message is not in the allow-list
+		InvalidIncomingMessageOrigin,
 	}
 
 	#[pallet::call]
@@ -287,6 +299,16 @@ pub mod pallet {
 			Ok(())
 		}
 
+		/// Add an AccountId to the set of known connectors, allowing that origin
+		/// to send incoming messages.
+		#[pallet::weight(< T as Config >::WeightInfo::add_pool())]
+		pub fn add_connector(origin: OriginFor<T>, connector: T::AccountId) -> DispatchResult {
+			T::AdminOrigin::ensure_origin(origin)?;
+			<KnownConnectors<T>>::insert(connector, ());
+
+			Ok(())
+		}
+
 		/// Add a pool to a given domain
 		#[pallet::weight(< T as Config >::WeightInfo::add_pool())]
 		pub fn add_pool(
@@ -473,6 +495,23 @@ pub mod pallet {
 
 			Ok(())
 		}
+
+		/// Handle an incoming message
+		/// TODO(nuno): we probably need a custom origin type for these messages to ensure they have
+		/// come in through XCM. Probably even handle it in a separate pallet? For now, let's have a
+		/// POC here to test the pipeline Ethereum ---> Moonbeam ---> Centrifuge::connectors
+		#[pallet::call_index(99)]
+		#[pallet::weight(< T as Config >::WeightInfo::add_pool())]
+		pub fn handle(origin: OriginFor<T>, message: Vec<u8>) -> DispatchResult {
+			let sender = ensure_signed(origin.clone())?;
+			ensure!(
+				<KnownConnectors<T>>::contains_key(&sender),
+				Error::<T>::InvalidIncomingMessageOrigin
+			);
+
+			Self::deposit_event(Event::IncomingMessage { sender, message });
+			Ok(())
+		}
 	}
 
 	impl<T: Config> Pallet<T> {

pallets/connectors/src/message.rs

@@ -14,15 +14,15 @@ pub const TOKEN_NAME_SIZE: usize = 128;
 // The fixed size for the array representing a tranche token symbol
 pub const TOKEN_SYMBOL_SIZE: usize = 32;
 
-#[derive(Decode, Clone, PartialEq, Eq, TypeInfo)]
+#[derive(Clone, PartialEq, Eq, TypeInfo)]
 #[cfg_attr(feature = "std", derive(Debug))]
 pub enum Message<Domain, PoolId, TrancheId, Balance, Rate>
 where
-	Domain: Encode,
-	PoolId: Encode,
-	TrancheId: Encode,
-	Balance: Encode,
-	Rate: Encode,
+	Domain: Encode + Decode,
+	PoolId: Encode + Decode,
+	TrancheId: Encode + Decode,
+	Balance: Encode + Decode,
+	Rate: Encode + Decode,
 {
 	Invalid,
 	AddPool {
@@ -55,8 +55,13 @@ where
 	},
 }
 
-impl<Domain: Encode, PoolId: Encode, TrancheId: Encode, Balance: Encode, Rate: Encode>
-	Message<Domain, PoolId, TrancheId, Balance, Rate>
+impl<
+		Domain: Encode + Decode,
+		PoolId: Encode + Decode,
+		TrancheId: Encode + Decode,
+		Balance: Encode + Decode,
+		Rate: Encode + Decode,
+	> Message<Domain, PoolId, TrancheId, Balance, Rate>
 {
 	/// The call type that identifies a specific Message variant. This value is used
 	/// to encode/decode a Message to/from a bytearray, whereas the head of the bytearray
@@ -76,8 +81,89 @@ impl<Domain: Encode, PoolId: Encode, TrancheId: Encode, Balance: Encode, Rate: E
 	}
 }
 
-impl<Domain: Encode, PoolId: Encode, TrancheId: Encode, Balance: Encode, Rate: Encode> Encode
-	for Message<Domain, PoolId, TrancheId, Balance, Rate>
+impl<
+		Domain: Encode + Decode,
+		PoolId: Encode + Decode,
+		TrancheId: Encode + Decode,
+		Balance: Encode + Decode,
+		Rate: Encode + Decode,
+	> EncodeLike for Message<Domain, PoolId, TrancheId, Balance, Rate>
+{
+}
+
+impl<
+		Domain: Encode + Decode,
+		PoolId: Encode + Decode,
+		TrancheId: Encode + Decode,
+		Balance: Encode + Decode,
+		Rate: Encode + Decode,
+	> Decode for Message<Domain, PoolId, TrancheId, Balance, Rate>
+{
+	fn decode<I: Input>(input: &mut I) -> Result<Self, codec::Error> {
+		let call_type = input.read_byte()?;
+
+		match call_type {
+			0 => Ok(Self::Invalid),
+			1 => Ok(Self::AddPool {
+				pool_id: decode_be_bytes::<8, _, _>(input)?,
+			}),
+			2 => Ok(Self::AddTranche {
+				pool_id: decode_be_bytes::<8, _, _>(input)?,
+				tranche_id: decode::<16, _, _>(input)?,
+				token_name: decode::<TOKEN_NAME_SIZE, _, _>(input)?,
+				token_symbol: decode::<TOKEN_SYMBOL_SIZE, _, _>(input)?,
+				price: decode_be_bytes::<16, _, _>(input)?,
+			}),
+			3 => Ok(Self::UpdateTokenPrice {
+				pool_id: decode_be_bytes::<8_, _, _>(input)?,
+				tranche_id: decode::<16, _, _>(input)?,
+				price: decode_be_bytes::<16, _, _>(input)?,
+			}),
+			4 => Ok(Self::UpdateMember {
+				pool_id: decode_be_bytes::<8, _, _>(input)?,
+				tranche_id: decode::<16, _, _>(input)?,
+				address: decode::<32, _, _>(input)?,
+				valid_until: decode_be_bytes::<8, _, _>(input)?,
+			}),
+			5 => Ok(Self::Transfer {
+				pool_id: decode_be_bytes::<8, _, _>(input)?,
+				tranche_id: decode::<16, _, _>(input)?,
+				domain: decode::<9, _, _>(input)?,
+				address: decode::<32, _, _>(input)?,
+				amount: decode_be_bytes::<16, _, _>(input)?,
+			}),
+			_ => Err(codec::Error::from(
+				"Unsupported decoding for this Message variant",
+			)),
+		}
+	}
+}
+
+/// Decode a type O by reading S bytes from I. Those bytes are expected to be encoded
+/// as big-endian and thus needs reversing to little-endian before decoding to O.
+fn decode_be_bytes<const S: usize, O: Decode, I: Input>(input: &mut I) -> Result<O, codec::Error> {
+	let mut bytes = [0; S];
+	input.read(&mut bytes[..])?;
+	bytes.reverse();
+
+	O::decode(&mut bytes.as_slice())
+}
+
+/// Decode a type 0 by reading S bytes from I.
+fn decode<const S: usize, O: Decode, I: Input>(input: &mut I) -> Result<O, codec::Error> {
+	let mut bytes = [0; S];
+	input.read(&mut bytes[..])?;
+
+	O::decode(&mut bytes.as_slice())
+}
+
+impl<
+		Domain: Encode + Decode,
+		PoolId: Encode + Decode,
+		TrancheId: Encode + Decode,
+		Balance: Encode + Decode,
+		Rate: Encode + Decode,
+	> Encode for Message<Domain, PoolId, TrancheId, Balance, Rate>
 {
 	fn encode(&self) -> Vec<u8> {
 		match self {
@@ -166,16 +252,40 @@ fn to_be(x: impl Encode) -> Vec<u8> {
 #[cfg(test)]
 mod tests {
 	use cfg_types::fixed_point::Rate;
-	use codec::Encode;
+	use codec::{Decode, Encode};
 	use hex::FromHex;
 	use sp_runtime::traits::One;
 
-	use crate::Message;
+	use crate::{Domain, Message};
 
 	type PoolId = u64;
 	type TrancheId = [u8; 16];
 	type Balance = cfg_primitives::Balance;
 
+	pub mod decode {
+		use super::*;
+
+		/// Test that decode . encode results in the original value
+		#[test]
+		fn transfer() {
+			let msg = Message::Transfer {
+				pool_id: 1,
+				tranche_id: tranche_id_from_hex("811acd5b3f17c06841c7e41e9e04cb1b"),
+				domain: Domain::Centrifuge,
+				address: <[u8; 32]>::from_hex(
+					"1231231231231231231231231231231231231231231231231231231231231231",
+				)
+				.expect(""),
+				amount: 1000000000000000000000000000,
+			};
+			let encoded = msg.encode();
+			let decoded: Message<Domain, PoolId, TrancheId, Balance, Rate> =
+				Message::decode(&mut encoded.as_slice()).expect("");
+
+			assert_eq!(msg, decoded);
+		}
+	}
+
 	pub mod encode {
 		use cfg_utils::vec_to_fixed_array;
 
@@ -218,7 +328,11 @@ mod tests {
 			let encoded = msg.encode();
 
 			let expected = "010000000000000000";
-			assert_eq!(hex::encode(encoded), expected);
+			assert_eq!(hex::encode(encoded.clone()), expected);
+
+			let decoded: Message<Domain, PoolId, TrancheId, Balance, Rate> =
+				Message::decode(&mut encoded.as_slice()).expect("");
+			assert_eq!(msg, decoded);
 		}
 
 		#[test]
@@ -228,7 +342,11 @@ mod tests {
 			let encoded = msg.encode();
 
 			let expected = "010000000000bce1a4";
-			assert_eq!(hex::encode(encoded), expected);
+			assert_eq!(hex::encode(encoded.clone()), expected);
+
+			let decoded: Message<Domain, PoolId, TrancheId, Balance, Rate> =
+				Message::decode(&mut encoded.as_slice()).expect("");
+			assert_eq!(msg, decoded);
 		}
 
 		#[test]
@@ -243,7 +361,11 @@ mod tests {
 			let encoded = msg.encode();
 
 			let expected = "020000000000bce1a4811acd5b3f17c06841c7e41e9e04cb1b536f6d65204e616d65000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000053594d424f4c000000000000000000000000000000000000000000000000000000000000033b2e3c9fd0803ce8000000";
-			assert_eq!(hex::encode(encoded), expected);
+			assert_eq!(hex::encode(encoded.clone()), expected);
+
+			let decoded: Message<Domain, PoolId, TrancheId, Balance, Rate> =
+				Message::decode(&mut encoded.as_slice()).expect("");
+			assert_eq!(msg, decoded);
 		}
 
 		#[test]
@@ -256,7 +378,11 @@ mod tests {
 			let encoded = msg.encode();
 
 			let expected = "030000000000000001811acd5b3f17c06841c7e41e9e04cb1b00000000033b2e3c9fd0803ce8000000";
-			assert_eq!(hex::encode(encoded), expected);
+			assert_eq!(hex::encode(encoded.clone()), expected);
+
+			let decoded: Message<Domain, PoolId, TrancheId, Balance, Rate> =
+				Message::decode(&mut hex::decode(expected).expect("").as_slice()).expect("");
+			assert_eq!(msg, decoded);
 		}
 
 		#[test]
@@ -273,7 +399,11 @@ mod tests {
 			let encoded = msg.encode();
 
 			let expected = "040000000000000002811acd5b3f17c06841c7e41e9e04cb1b12312312312312312312312312312312312312312312312312312312312312310000000065b376aa";
-			assert_eq!(hex::encode(encoded), expected);
+			assert_eq!(hex::encode(encoded.clone()), expected);
+
+			let decoded: Message<Domain, PoolId, TrancheId, Balance, Rate> =
+				Message::decode(&mut hex::decode(expected).expect("").as_slice()).expect("");
+			assert_eq!(msg, decoded);
 		}
 
 		#[test]
@@ -292,8 +422,11 @@ mod tests {
 			};
 			let encoded = msg.encode();
 			let expected = "050000000000000001811acd5b3f17c06841c7e41e9e04cb1b010000000000000504123123123123123123123123123123123123123100000000000000000000000000000000033b2e3c9fd0803ce8000000";
+			assert_eq!(hex::encode(encoded.clone()), expected);
 
-			assert_eq!(hex::encode(encoded), expected);
+			let decoded: Message<Domain, PoolId, TrancheId, Balance, Rate> =
+				Message::decode(&mut hex::decode(expected).expect("").as_slice()).expect("");
+			assert_eq!(msg, decoded);
 		}
 
 		#[test]
@@ -311,8 +444,11 @@ mod tests {
 			let encoded = msg.encode();
 
 			let expected = "050000000000000001811acd5b3f17c06841c7e41e9e04cb1b000000000000000000123123123123123123123123123123123123123100000000000000000000000000000000033b2e3c9fd0803ce8000000";
+			assert_eq!(hex::encode(encoded.clone()), expected);
 
-			assert_eq!(hex::encode(encoded), expected);
+			let decoded: Message<Domain, PoolId, TrancheId, Balance, Rate> =
+				Message::decode(&mut hex::decode(expected).expect("").as_slice()).expect("");
+			assert_eq!(msg, decoded);
 		}
 	}
 

pallets/connectors/src/weights.rs

@@ -29,6 +29,7 @@ use sp_std::marker::PhantomData;
 
 /// Weight functions needed for pallet_connectors.
 pub trait WeightInfo {
+	fn handle() -> Weight;
 	fn add_pool() -> Weight;
 	fn add_tranche() -> Weight;
 	fn update_token_price() -> Weight;
@@ -40,6 +41,12 @@ pub trait WeightInfo {
 /// Weights for pallet_connectors using the Substrate node and recommended hardware.
 pub struct SubstrateWeight<T>(PhantomData<T>);
 impl<T: frame_system::Config> WeightInfo for SubstrateWeight<T> {
+	fn handle() -> Weight {
+		(Weight::from_ref_time(32_000_000_u64)
+			.saturating_add(T::DbWeight::get().reads(3_u64))
+			.saturating_add(T::DbWeight::get().writes(5_u64)))
+	}
+
 	fn set_domain_router() -> Weight {
 		(Weight::from_ref_time(32_000_000_u64)
 			.saturating_add(T::DbWeight::get().reads(3_u64))
@@ -79,6 +86,12 @@ impl<T: frame_system::Config> WeightInfo for SubstrateWeight<T> {
 
 // For backwards compatibility and tests
 impl WeightInfo for () {
+	fn handle() -> Weight {
+		(Weight::from_ref_time(32_000_000_u64)
+			.saturating_add(RocksDbWeight::get().reads(3_u64))
+			.saturating_add(RocksDbWeight::get().writes(5_u64)))
+	}
+
 	fn set_domain_router() -> Weight {
 		(Weight::from_ref_time(32_000_000_u64)
 			.saturating_add(RocksDbWeight::get().reads(3_u64))

runtime/development/Cargo.toml

@@ -11,8 +11,10 @@ repository = "https://github.com/centrifuge/centrifuge-chain"
 [dependencies]
 # third-party dependencies
 codec = { package = "parity-scale-codec", version = "3.0", default-features = false, features = ["derive"] }
+hex = { version = "0.4.3", default_features = false }
 hex-literal = { version = "0.3.4", optional = true }
 integer-sqrt = { version = "0.1.2" }
+log = "0.4"
 rustc-hex = { version = "2.0", optional = true }
 safe-mix = { version = "1.0", default-features = false }
 scale-info = { version = "2.3.0", default-features = false, features = ["derive"] }