Substrate relay guards (#470)

* substrate relay guards

* checked time condition

* ChainWithBalances

* removed obsolete comment

* Update relays/substrate-client/src/chain.rs

Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com>

* trailing space

Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com>
This commit is contained in:
Svyatoslav Nikolsky
2020-11-02 15:01:59 +03:00
committed by Bastian Köcher
parent cc1da1bb39
commit b027c81266
12 changed files with 498 additions and 20 deletions
+1
View File
@@ -17,6 +17,7 @@ millau-runtime = { path = "../../bin/millau/runtime" }
# Substrate Dependencies
frame-support = "2.0"
frame-system = "2.0"
pallet-transaction-payment = "2.0"
sp-core = "2.0"
+17 -2
View File
@@ -18,12 +18,13 @@
use codec::Encode;
use headers_relay::sync_types::SourceHeader;
use relay_substrate_client::{Chain, ChainBase, Client, TransactionSignScheme};
use sp_core::Pair;
use relay_substrate_client::{Chain, ChainBase, ChainWithBalances, Client, TransactionSignScheme};
use sp_core::{storage::StorageKey, Pair};
use sp_runtime::{
generic::SignedPayload,
traits::{Header as HeaderT, IdentifyAccount},
};
use std::time::Duration;
/// Millau header id.
pub type HeaderId = relay_utils::HeaderId<millau_runtime::Hash, millau_runtime::BlockNumber>;
@@ -40,12 +41,26 @@ impl ChainBase for Millau {
}
impl Chain for Millau {
const NAME: &'static str = "Millau";
const AVERAGE_BLOCK_INTERVAL: Duration = Duration::from_secs(5);
type AccountId = millau_runtime::AccountId;
type Index = millau_runtime::Index;
type SignedBlock = millau_runtime::SignedBlock;
type Call = millau_runtime::Call;
}
impl ChainWithBalances for Millau {
type NativeBalance = millau_runtime::Balance;
fn account_info_storage_key(account_id: &Self::AccountId) -> StorageKey {
use frame_support::storage::generator::StorageMap;
StorageKey(frame_system::Account::<millau_runtime::Runtime>::storage_map_final_key(
account_id,
))
}
}
impl TransactionSignScheme for Millau {
type Chain = Millau;
type AccountKeyPair = sp_core::sr25519::Pair;
+1
View File
@@ -18,6 +18,7 @@ rialto-runtime = { path = "../../bin/rialto/runtime" }
# Substrate Dependencies
frame-system = "2.0"
frame-support = "2.0"
pallet-transaction-payment = "2.0"
sp-core = "2.0"
sp-keyring = "2.0"
+17 -2
View File
@@ -18,12 +18,13 @@
use codec::Encode;
use headers_relay::sync_types::SourceHeader;
use relay_substrate_client::{Chain, ChainBase, Client, TransactionSignScheme};
use sp_core::Pair;
use relay_substrate_client::{Chain, ChainBase, ChainWithBalances, Client, TransactionSignScheme};
use sp_core::{storage::StorageKey, Pair};
use sp_runtime::{
generic::SignedPayload,
traits::{Header as HeaderT, IdentifyAccount},
};
use std::time::Duration;
pub use rialto_runtime::BridgeMillauCall;
@@ -42,12 +43,26 @@ impl ChainBase for Rialto {
}
impl Chain for Rialto {
const NAME: &'static str = "Rialto";
const AVERAGE_BLOCK_INTERVAL: Duration = Duration::from_secs(5);
type AccountId = rialto_runtime::AccountId;
type Index = rialto_runtime::Index;
type SignedBlock = rialto_runtime::SignedBlock;
type Call = rialto_runtime::Call;
}
impl ChainWithBalances for Rialto {
type NativeBalance = rialto_runtime::Balance;
fn account_info_storage_key(account_id: &Self::AccountId) -> StorageKey {
use frame_support::storage::generator::StorageMap;
StorageKey(frame_system::Account::<rialto_runtime::Runtime>::storage_map_final_key(
account_id,
))
}
}
impl TransactionSignScheme for Rialto {
type Chain = Rialto;
type AccountKeyPair = sp_core::sr25519::Pair;
+7 -1
View File
@@ -6,11 +6,13 @@ edition = "2018"
license = "GPL-3.0-or-later WITH Classpath-exception-2.0"
[dependencies]
async-std = "1.6.5"
async-trait = "0.1.40"
codec = { package = "parity-scale-codec", version = "1.3.4" }
jsonrpsee = { git = "https://github.com/svyatonik/jsonrpsee.git", branch = "shared-client-in-rpc-api", default-features = false, features = ["ws"] }
log = "0.4.11"
num-traits = "0.2"
rand = "0.7"
# Bridge dependencies
@@ -22,6 +24,10 @@ relay-utils = { path = "../utils" }
frame-support = "2.0"
frame-system = "2.0"
pallet-balances = "2.0"
sp-core = "2.0"
sp-runtime = "2.0"
sp-std = "2.0"
sp-version = "2.0"
#[dev-dependencies]
futures = "0.3.7"
+26 -2
View File
@@ -19,16 +19,24 @@ use crate::client::Client;
use bp_runtime::Chain as ChainBase;
use frame_support::Parameter;
use jsonrpsee::common::{DeserializeOwned, Serialize};
use sp_core::Pair;
use num_traits::{CheckedSub, Zero};
use sp_core::{storage::StorageKey, Pair};
use sp_runtime::{
generic::SignedBlock,
traits::{AtLeast32Bit, Dispatchable, MaybeDisplay, MaybeSerialize, MaybeSerializeDeserialize, Member},
Justification,
};
use sp_std::fmt::Debug;
use std::{fmt::Debug, time::Duration};
/// Substrate-based chain from minimal relay-client point of view.
pub trait Chain: ChainBase {
/// Chain name.
const NAME: &'static str;
/// Average block interval.
///
/// How often blocks are produced on that chain. It's suggested to set this value to match the block time of the chain.
const AVERAGE_BLOCK_INTERVAL: Duration;
/// The user account identifier type for the runtime.
type AccountId: Parameter + Member + MaybeSerializeDeserialize + Debug + MaybeDisplay + Ord + Default;
/// Account index (aka nonce) type. This stores the number of previous transactions associated
@@ -40,6 +48,16 @@ pub trait Chain: ChainBase {
type Call: Dispatchable + Debug;
}
/// Substrate-based chain with `frame_system::Trait::AccountData` set to
/// the `pallet_balances::AccountData<NativeBalance>`.
pub trait ChainWithBalances: Chain {
/// Balance of an account in native tokens.
type NativeBalance: Parameter + Member + DeserializeOwned + Clone + Copy + CheckedSub + PartialOrd + Zero;
/// Return runtime storage key for getting `frame_system::AccountInfo` of given account.
fn account_info_storage_key(account_id: &Self::AccountId) -> StorageKey;
}
/// Block with justification.
pub trait BlockWithJustification {
/// Return block justification, if known.
@@ -64,6 +82,12 @@ pub trait TransactionSignScheme {
) -> Self::SignedTransaction;
}
impl BlockWithJustification for () {
fn justification(&self) -> Option<&Justification> {
None
}
}
impl<Block> BlockWithJustification for SignedBlock<Block> {
fn justification(&self) -> Option<&Justification> {
self.justification.as_ref()
+43 -10
View File
@@ -16,16 +16,21 @@
//! Substrate node client.
use crate::chain::Chain;
use crate::chain::{Chain, ChainWithBalances};
use crate::error::Error;
use crate::rpc::Substrate;
use crate::{ConnectionParams, Result};
use codec::Decode;
use frame_system::AccountInfo;
use jsonrpsee::common::DeserializeOwned;
use jsonrpsee::raw::RawClient;
use jsonrpsee::transport::ws::WsTransportClient;
use jsonrpsee::{client::Subscription, Client as RpcClient};
use num_traits::Zero;
use pallet_balances::AccountData;
use sp_core::Bytes;
use sp_version::RuntimeVersion;
const SUB_API_GRANDPA_AUTHORITIES: &str = "GrandpaApi_grandpa_authorities";
@@ -69,18 +74,17 @@ impl<C: Chain> Client<C> {
}
}
impl<C: Chain> Client<C>
where
C::Header: DeserializeOwned,
C::Index: DeserializeOwned,
{
impl<C: Chain> Client<C> {
/// Return hash of the genesis block.
pub fn genesis_hash(&self) -> &C::Hash {
&self.genesis_hash
}
/// Returns the best Substrate header.
pub async fn best_header(&self) -> Result<C::Header> {
pub async fn best_header(&self) -> Result<C::Header>
where
C::Header: DeserializeOwned,
{
Ok(Substrate::<C, _, _>::chain_get_header(&self.client, None).await?)
}
@@ -90,7 +94,10 @@ where
}
/// Get a Substrate header by its hash.
pub async fn header_by_hash(&self, block_hash: C::Hash) -> Result<C::Header> {
pub async fn header_by_hash(&self, block_hash: C::Hash) -> Result<C::Header>
where
C::Header: DeserializeOwned,
{
Ok(Substrate::<C, _, _>::chain_get_header(&self.client, block_hash).await?)
}
@@ -100,15 +107,41 @@ where
}
/// Get a Substrate header by its number.
pub async fn header_by_number(&self, block_number: C::BlockNumber) -> Result<C::Header> {
pub async fn header_by_number(&self, block_number: C::BlockNumber) -> Result<C::Header>
where
C::Header: DeserializeOwned,
{
let block_hash = Self::block_hash_by_number(self, block_number).await?;
Ok(Self::header_by_hash(self, block_hash).await?)
}
/// Return runtime version.
pub async fn runtime_version(&self) -> Result<RuntimeVersion> {
Ok(Substrate::<C, _, _>::runtime_version(&self.client).await?)
}
/// Return native tokens balance of the account.
pub async fn free_native_balance(&self, account: C::AccountId) -> Result<C::NativeBalance>
where
C: ChainWithBalances,
{
let storage_key = C::account_info_storage_key(&account);
let encoded_account_data = Substrate::<C, _, _>::get_storage(&self.client, storage_key)
.await?
.ok_or(Error::AccountDoesNotExist)?;
let decoded_account_data =
AccountInfo::<C::Index, AccountData<C::NativeBalance>>::decode(&mut &encoded_account_data.0[..])
.map_err(Error::ResponseParseFailed)?;
Ok(decoded_account_data.data.free)
}
/// Get the nonce of the given Substrate account.
///
/// Note: It's the caller's responsibility to make sure `account` is a valid ss58 address.
pub async fn next_account_index(&self, account: C::AccountId) -> Result<C::Index> {
pub async fn next_account_index(&self, account: C::AccountId) -> Result<C::Index>
where
C::Index: DeserializeOwned,
{
Ok(Substrate::<C, _, _>::system_account_next_index(&self.client, account).await?)
}
@@ -34,6 +34,8 @@ pub enum Error {
Request(RequestError),
/// The response from the server could not be SCALE decoded.
ResponseParseFailed(codec::Error),
/// Account does not exist on the chain.
AccountDoesNotExist,
}
impl From<WsNewDnsError> for Error {
@@ -66,6 +68,7 @@ impl ToString for Error {
Self::WsConnectionError(e) => e.to_string(),
Self::Request(e) => e.to_string(),
Self::ResponseParseFailed(e) => e.what().to_string(),
Self::AccountDoesNotExist => "Account does not exist on the chain".into(),
}
}
}
@@ -0,0 +1,371 @@
// Copyright 2019-2020 Parity Technologies (UK) Ltd.
// This file is part of Parity Bridges Common.
// Parity Bridges Common is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// Parity Bridges Common is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with Parity Bridges Common. If not, see <http://www.gnu.org/licenses/>.
//! Module provides a set of guard functions that are running in background threads
//! and are aborting process if some condition fails.
use crate::{Chain, ChainWithBalances, Client};
use async_trait::async_trait;
use num_traits::CheckedSub;
use sp_version::RuntimeVersion;
use std::{
collections::VecDeque,
time::{Duration, Instant},
};
/// Guards environment.
#[async_trait]
pub trait Environment<C: ChainWithBalances>: Send + Sync + 'static {
/// Return current runtime version.
async fn runtime_version(&mut self) -> Result<RuntimeVersion, String>;
/// Return free native balance of the account on the chain.
async fn free_native_balance(&mut self, account: C::AccountId) -> Result<C::NativeBalance, String>;
/// Return current time.
fn now(&self) -> Instant {
Instant::now()
}
/// Sleep given amount of time.
async fn sleep(&mut self, duration: Duration) {
async_std::task::sleep(duration).await
}
/// Abort current process. Called when guard condition check fails.
async fn abort(&mut self) {
std::process::abort();
}
}
/// Abort when runtime spec version is different from specified.
pub fn abort_on_spec_version_change<C: ChainWithBalances>(mut env: impl Environment<C>, expected_spec_version: u32) {
async_std::task::spawn(async move {
loop {
let actual_spec_version = env.runtime_version().await;
match actual_spec_version {
Ok(version) if version.spec_version == expected_spec_version => (),
Ok(version) => {
log::error!(
target: "bridge-guard",
"{} runtime spec version has changed from {} to {}. Aborting relay",
C::NAME,
expected_spec_version,
version.spec_version,
);
env.abort().await;
}
Err(error) => log::warn!(
target: "bridge-guard",
"Failed to read {} runtime version: {:?}. Relay may need to be stopped manually",
C::NAME,
error,
),
}
env.sleep(conditions_check_delay::<C>()).await;
}
});
}
/// Abort if, during a 24 hours, free balance of given account is decreased at least by given value.
/// Other components may increase (or decrease) balance of account and it WILL affect logic of the guard.
pub fn abort_when_account_balance_decreased<C: ChainWithBalances>(
mut env: impl Environment<C>,
account_id: C::AccountId,
maximal_decrease: C::NativeBalance,
) {
const DAY: Duration = Duration::from_secs(60 * 60 * 24);
async_std::task::spawn(async move {
let mut balances = VecDeque::new();
loop {
let current_time = env.now();
// remember balances that are beyound 24h border
let time_border = current_time - DAY;
while balances.front().map(|(time, _)| *time < time_border).unwrap_or(false) {
balances.pop_front();
}
// read balance of the account
let current_balance = env.free_native_balance(account_id.clone()).await;
// remember balance and check difference
match current_balance {
Ok(current_balance) => {
// remember balance
balances.push_back((current_time, current_balance));
// check if difference between current and oldest balance is too large
let (oldest_time, oldest_balance) =
balances.front().expect("pushed to queue couple of lines above; qed");
let balances_difference = oldest_balance.checked_sub(&current_balance);
if balances_difference > Some(maximal_decrease) {
log::error!(
target: "bridge-guard",
"Balance of {} account {:?} has decreased from {:?} to {:?} in {} minutes. Aborting relay",
C::NAME,
account_id,
oldest_balance,
current_balance,
current_time.duration_since(*oldest_time).as_secs() / 60,
);
env.abort().await;
}
}
Err(error) => {
log::warn!(
target: "bridge-guard",
"Failed to read {} account {:?} balance: {:?}. Relay may need to be stopped manually",
C::NAME,
account_id,
error,
);
}
};
env.sleep(conditions_check_delay::<C>()).await;
}
});
}
/// Delay between conditions check.
fn conditions_check_delay<C: Chain>() -> Duration {
C::AVERAGE_BLOCK_INTERVAL * (10 + rand::random::<u32>() % 10)
}
#[async_trait]
impl<C: ChainWithBalances> Environment<C> for Client<C> {
async fn runtime_version(&mut self) -> Result<RuntimeVersion, String> {
Client::<C>::runtime_version(self).await.map_err(|e| e.to_string())
}
async fn free_native_balance(&mut self, account: C::AccountId) -> Result<C::NativeBalance, String> {
Client::<C>::free_native_balance(self, account)
.await
.map_err(|e| e.to_string())
}
}
#[cfg(test)]
mod tests {
use super::*;
use futures::{
channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender},
future::FutureExt,
stream::StreamExt,
SinkExt,
};
struct TestChain;
impl bp_runtime::Chain for TestChain {
type BlockNumber = u32;
type Hash = sp_core::H256;
type Hasher = sp_runtime::traits::BlakeTwo256;
type Header = sp_runtime::generic::Header<u32, sp_runtime::traits::BlakeTwo256>;
}
impl Chain for TestChain {
const NAME: &'static str = "Test";
const AVERAGE_BLOCK_INTERVAL: Duration = Duration::from_millis(1);
type AccountId = u32;
type Index = u32;
type SignedBlock = ();
type Call = ();
}
impl ChainWithBalances for TestChain {
type NativeBalance = u32;
fn account_info_storage_key(_account_id: &u32) -> sp_core::storage::StorageKey {
unreachable!()
}
}
struct TestEnvironment {
runtime_version_rx: UnboundedReceiver<RuntimeVersion>,
free_native_balance_rx: UnboundedReceiver<u32>,
slept_tx: UnboundedSender<()>,
aborted_tx: UnboundedSender<()>,
}
#[async_trait]
impl Environment<TestChain> for TestEnvironment {
async fn runtime_version(&mut self) -> Result<RuntimeVersion, String> {
Ok(self.runtime_version_rx.next().await.unwrap_or_default())
}
async fn free_native_balance(&mut self, _account: u32) -> Result<u32, String> {
Ok(self.free_native_balance_rx.next().await.unwrap_or_default())
}
async fn sleep(&mut self, _duration: Duration) {
let _ = self.slept_tx.send(()).await;
}
async fn abort(&mut self) {
let _ = self.aborted_tx.send(()).await;
// simulate process abort :)
async_std::task::sleep(Duration::from_secs(60)).await;
}
}
#[test]
fn aborts_when_spec_version_is_changed() {
async_std::task::block_on(async {
let (
(mut runtime_version_tx, runtime_version_rx),
(_free_native_balance_tx, free_native_balance_rx),
(slept_tx, mut slept_rx),
(aborted_tx, mut aborted_rx),
) = (unbounded(), unbounded(), unbounded(), unbounded());
abort_on_spec_version_change(
TestEnvironment {
runtime_version_rx,
free_native_balance_rx,
slept_tx,
aborted_tx,
},
0,
);
// client responds with wrong version
runtime_version_tx
.send(RuntimeVersion {
spec_version: 42,
..Default::default()
})
.await
.unwrap();
// then the `abort` function is called
aborted_rx.next().await;
// and we do not reach the `sleep` function call
assert!(slept_rx.next().now_or_never().is_none());
});
}
#[test]
fn does_not_aborts_when_spec_version_is_unchanged() {
async_std::task::block_on(async {
let (
(mut runtime_version_tx, runtime_version_rx),
(_free_native_balance_tx, free_native_balance_rx),
(slept_tx, mut slept_rx),
(aborted_tx, mut aborted_rx),
) = (unbounded(), unbounded(), unbounded(), unbounded());
abort_on_spec_version_change(
TestEnvironment {
runtime_version_rx,
free_native_balance_rx,
slept_tx,
aborted_tx,
},
42,
);
// client responds with the same version
runtime_version_tx
.send(RuntimeVersion {
spec_version: 42,
..Default::default()
})
.await
.unwrap();
// then the `sleep` function is called
slept_rx.next().await;
// and the `abort` function is not called
assert!(aborted_rx.next().now_or_never().is_none());
});
}
#[test]
fn aborts_when_balance_is_too_low() {
async_std::task::block_on(async {
let (
(_runtime_version_tx, runtime_version_rx),
(mut free_native_balance_tx, free_native_balance_rx),
(slept_tx, mut slept_rx),
(aborted_tx, mut aborted_rx),
) = (unbounded(), unbounded(), unbounded(), unbounded());
abort_when_account_balance_decreased(
TestEnvironment {
runtime_version_rx,
free_native_balance_rx,
slept_tx,
aborted_tx,
},
0,
100,
);
// client responds with initial balance
free_native_balance_tx.send(1000).await.unwrap();
// then the guard sleeps
slept_rx.next().await;
// and then client responds with updated balance, which is too low
free_native_balance_tx.send(899).await.unwrap();
// then the `abort` function is called
aborted_rx.next().await;
// and we do not reach next `sleep` function call
assert!(slept_rx.next().now_or_never().is_none());
});
}
#[test]
fn does_not_aborts_when_balance_is_enough() {
async_std::task::block_on(async {
let (
(_runtime_version_tx, runtime_version_rx),
(mut free_native_balance_tx, free_native_balance_rx),
(slept_tx, mut slept_rx),
(aborted_tx, mut aborted_rx),
) = (unbounded(), unbounded(), unbounded(), unbounded());
abort_when_account_balance_decreased(
TestEnvironment {
runtime_version_rx,
free_native_balance_rx,
slept_tx,
aborted_tx,
},
0,
100,
);
// client responds with initial balance
free_native_balance_tx.send(1000).await.unwrap();
// then the guard sleeps
slept_rx.next().await;
// and then client responds with updated balance, which is enough
free_native_balance_tx.send(950).await.unwrap();
// then the `sleep` function is called
slept_rx.next().await;
// and `abort` is not called
assert!(aborted_rx.next().now_or_never().is_none());
});
}
}
+2 -1
View File
@@ -23,9 +23,10 @@ mod client;
mod error;
mod rpc;
pub mod guard;
pub mod headers_source;
pub use crate::chain::{BlockWithJustification, Chain, TransactionSignScheme};
pub use crate::chain::{BlockWithJustification, Chain, ChainWithBalances, TransactionSignScheme};
pub use crate::client::{Client, JustificationsSubscription, OpaqueGrandpaAuthoritiesSet};
pub use crate::error::{Error, Result};
pub use bp_runtime::{BlockNumberOf, Chain as ChainBase, HashOf, HeaderOf};
+9 -1
View File
@@ -23,7 +23,11 @@
use crate::chain::Chain;
use sp_core::Bytes;
use sp_core::{
storage::{StorageData, StorageKey},
Bytes,
};
use sp_version::RuntimeVersion;
jsonrpsee::rpc_api! {
pub(crate) Substrate<C: Chain> {
@@ -39,5 +43,9 @@ jsonrpsee::rpc_api! {
fn author_submit_extrinsic(extrinsic: Bytes) -> C::Hash;
#[rpc(method = "state_call", positional_params)]
fn state_call(method: String, data: Bytes, at_block: Option<C::Hash>) -> Bytes;
#[rpc(method = "state_getStorage", positional_params)]
fn get_storage(key: StorageKey) -> Option<StorageData>;
#[rpc(method = "state_getRuntimeVersion", positional_params)]
fn runtime_version() -> RuntimeVersion;
}
}