Transaction Queue banning (#2524)

* Blacklisting transaction queue

* Using blacklisting queue in miner

* Restoring todo [ci:skip]

* Blacklisting recipients and code

* Renaming blacklisting->banning

* CLI option for banning.

* Fixing submodule commit [ci:skip]

* Fixing RPC tests

* Additional logging when dropping transactions

* whitespace

[ci:skip]

* Configurable ban duration

* Reverting fix for pruning history from config file

ethcore/Cargo.toml

@@ -24,6 +24,10 @@ rayon = "0.4.2"
 semver = "0.2"
 bit-set = "0.4"
 time = "0.1"
+rand = "0.3"
+byteorder = "0.5"
+transient-hashmap = "0.1"
+lru-cache = { git = "https://github.com/contain-rs/lru-cache" }
 evmjit = { path = "../evmjit", optional = true }
 clippy = { version = "0.0.96", optional = true}
 ethash = { path = "../ethash" }
@@ -36,10 +40,7 @@ ethstore = { path = "../ethstore" }
 ethkey = { path = "../ethkey" }
 ethcore-ipc-nano = { path = "../ipc/nano" }
 rlp = { path = "../util/rlp" }
-rand = "0.3"
-lru-cache = { git = "https://github.com/contain-rs/lru-cache" }
 ethcore-bloom-journal = { path = "../util/bloom" }
-byteorder = "0.5"
 
 [dependencies.hyper]
 git = "https://github.com/ethcore/hyper"

ethcore/src/error.rs

@@ -63,6 +63,12 @@ pub enum TransactionError {
 	},
 	/// Transaction's gas limit (aka gas) is invalid.
 	InvalidGasLimit(OutOfBounds<U256>),
+	/// Transaction sender is banned.
+	SenderBanned,
+	/// Transaction receipient is banned.
+	RecipientBanned,
+	/// Contract creation code is banned.
+	CodeBanned,
 }
 
 impl fmt::Display for TransactionError {
@@ -81,6 +87,9 @@ impl fmt::Display for TransactionError {
 			GasLimitExceeded { limit, got } =>
 				format!("Gas limit exceeded. Limit={}, Given={}", limit, got),
 			InvalidGasLimit(ref err) => format!("Invalid gas limit. {}", err),
+			SenderBanned => "Sender is temporarily banned.".into(),
+			RecipientBanned => "Recipient is temporarily banned.".into(),
+			CodeBanned => "Contract code is temporarily banned.".into(),
 		};
 
 		f.write_fmt(format_args!("Transaction error ({})", msg))

ethcore/src/lib.rs

@@ -101,6 +101,7 @@ extern crate bit_set;
 extern crate rlp;
 extern crate ethcore_bloom_journal as bloom_journal;
 extern crate byteorder;
+extern crate transient_hashmap;
 
 #[macro_use]
 extern crate log;

ethcore/src/miner/banning_queue.rs

@@ -0,0 +1,331 @@
+// Copyright 2015, 2016 Ethcore (UK) Ltd.
+// This file is part of Parity.
+
+// Parity 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 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.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Banning Queue
+//! Transacton Queue wrapper maintaining additional list of banned senders and contract hashes.
+
+use std::time::Duration;
+use std::ops::{Deref, DerefMut};
+use std::cell::Cell;
+use transaction::{SignedTransaction, Action};
+use transient_hashmap::TransientHashMap;
+use miner::{TransactionQueue, TransactionImportResult, TransactionOrigin, AccountDetails};
+use error::{Error, TransactionError};
+use util::{Uint, U256, H256, Address, Hashable};
+
+type Count = u16;
+
+/// Auto-Banning threshold
+pub enum Threshold {
+	/// Should ban after given number of misbehaves reported.
+	BanAfter(Count),
+	/// Should never ban anything
+	NeverBan
+}
+
+impl Default for Threshold {
+	fn default() -> Self {
+		Threshold::NeverBan
+	}
+}
+
+/// Transaction queue with banlist.
+pub struct BanningTransactionQueue {
+	queue: TransactionQueue,
+	ban_threshold: Threshold,
+	senders_bans: TransientHashMap<Address, Cell<Count>>,
+	recipients_bans: TransientHashMap<Address, Cell<Count>>,
+	codes_bans: TransientHashMap<H256, Cell<Count>>,
+}
+
+impl BanningTransactionQueue {
+	/// Creates new banlisting transaction queue
+	pub fn new(queue: TransactionQueue, ban_threshold: Threshold, ban_lifetime: Duration) -> Self {
+		let ban_lifetime_sec = ban_lifetime.as_secs();
+		assert!(ban_lifetime_sec > 0, "Lifetime has to be specified in seconds.");
+		BanningTransactionQueue {
+			queue: queue,
+			ban_threshold: ban_threshold,
+			senders_bans: TransientHashMap::new(ban_lifetime_sec),
+			recipients_bans: TransientHashMap::new(ban_lifetime_sec),
+			codes_bans: TransientHashMap::new(ban_lifetime_sec),
+		}
+	}
+
+	/// Borrows internal queue.
+	/// NOTE: you can insert transactions to the queue even
+	/// if they would be rejected because of ban otherwise.
+	/// But probably you shouldn't.
+	pub fn queue(&mut self) -> &mut TransactionQueue {
+		&mut self.queue
+	}
+
+	/// Add to the queue taking bans into consideration.
+	/// May reject transaction because of the banlist.
+	pub fn add_with_banlist<F>(
+		&mut self,
+		transaction: SignedTransaction,
+		account_details: &F,
+	) -> Result<TransactionImportResult, Error> where F: Fn(&Address) -> AccountDetails {
+		if let Threshold::BanAfter(threshold) = self.ban_threshold {
+			// NOTE In all checks use direct query to avoid increasing ban timeout.
+
+			// Check sender
+			if let Ok(sender) = transaction.sender() {
+				let count = self.senders_bans.direct().get(&sender).map(|v| v.get()).unwrap_or(0);
+				if count > threshold {
+					debug!(target: "txqueue", "Ignoring transaction {:?} because sender is banned.", transaction.hash());
+					return Err(Error::Transaction(TransactionError::SenderBanned));
+				}
+			}
+
+			// Check recipient
+			if let Action::Call(recipient) = transaction.action {
+				let count = self.recipients_bans.direct().get(&recipient).map(|v| v.get()).unwrap_or(0);
+				if count > threshold {
+					debug!(target: "txqueue", "Ignoring transaction {:?} because recipient is banned.", transaction.hash());
+					return Err(Error::Transaction(TransactionError::RecipientBanned));
+				}
+			}
+
+			// Check code
+			if let Action::Create = transaction.action {
+				let code_hash = transaction.data.sha3();
+				let count = self.codes_bans.direct().get(&code_hash).map(|v| v.get()).unwrap_or(0);
+				if count > threshold {
+					debug!(target: "txqueue", "Ignoring transaction {:?} because code is banned.", transaction.hash());
+					return Err(Error::Transaction(TransactionError::CodeBanned));
+				}
+			}
+		}
+		self.queue.add(transaction, account_details, TransactionOrigin::External)
+	}
+
+	/// Ban transaction with given hash.
+	/// Transaction has to be in the queue.
+	///
+	/// Bans sender and recipient/code and returns `true` when any ban has reached threshold.
+	pub fn ban_transaction(&mut self, hash: &H256) -> bool {
+		let transaction = self.queue.find(hash);
+		match transaction {
+			Some(transaction) => {
+				let sender = transaction.sender().expect("Transaction is in queue, so the sender is already validated; qed");
+				// Ban sender
+				let sender_banned = self.ban_sender(sender);
+				// Ban recipient and codehash
+				let is_banned = sender_banned || match transaction.action {
+					Action::Call(recipient) => {
+						self.ban_recipient(recipient)
+					},
+					Action::Create => {
+						self.ban_codehash(transaction.data.sha3())
+					},
+				};
+				is_banned
+			},
+			None => false,
+		}
+	}
+
+	/// Ban given sender.
+	/// If bans threshold is reached all subsequent transactions from this sender will be rejected.
+	/// Reaching bans threshold also removes all existsing transaction from this sender that are already in the
+	/// queue.
+	fn ban_sender(&mut self, address: Address) -> bool {
+		let count = {
+			let mut count = self.senders_bans.entry(address).or_insert_with(|| Cell::new(0));
+			*count.get_mut() = count.get().saturating_add(1);
+			count.get()
+		};
+		match self.ban_threshold {
+			Threshold::BanAfter(threshold) if count > threshold => {
+				// Banlist the sender.
+				// Remove all transactions from the queue.
+				self.remove_all(address, !U256::zero());
+				true
+			},
+			_ => false
+		}
+	}
+
+	/// Ban given recipient.
+	/// If bans threshold is reached all subsequent transactions to this address will be rejected.
+	/// Returns true if bans threshold has been reached.
+	fn ban_recipient(&mut self, address: Address) -> bool {
+		let count = {
+			let mut count = self.recipients_bans.entry(address).or_insert_with(|| Cell::new(0));
+			*count.get_mut() = count.get().saturating_add(1);
+			count.get()
+		};
+		match self.ban_threshold {
+			// TODO [ToDr] Consider removing other transactions to the same recipient from the queue?
+			Threshold::BanAfter(threshold) if count > threshold => true,
+			_ => false
+		}
+	}
+
+
+	/// Ban given codehash.
+	/// If bans threshold is reached all subsequent transactions to contracts with this codehash will be rejected.
+	/// Returns true if bans threshold has been reached.
+	fn ban_codehash(&mut self, code_hash: H256) -> bool {
+		let mut count = self.codes_bans.entry(code_hash).or_insert_with(|| Cell::new(0));
+		*count.get_mut() = count.get().saturating_add(1);
+
+		match self.ban_threshold {
+			// TODO [ToDr] Consider removing other transactions with the same code from the queue?
+			Threshold::BanAfter(threshold) if count.get() > threshold => true,
+			_ => false,
+		}
+	}
+}
+
+impl Deref for BanningTransactionQueue {
+	type Target = TransactionQueue;
+
+	fn deref(&self) -> &Self::Target {
+		&self.queue
+	}
+}
+impl DerefMut for BanningTransactionQueue {
+	fn deref_mut(&mut self) -> &mut Self::Target {
+		self.queue()
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use std::time::Duration;
+	use super::{BanningTransactionQueue, Threshold};
+	use ethkey::{Random, Generator};
+	use transaction::{Transaction, SignedTransaction, Action};
+	use error::{Error, TransactionError};
+	use client::TransactionImportResult;
+	use miner::{TransactionQueue, TransactionOrigin, AccountDetails};
+	use util::{Uint, U256, Address, FromHex, Hashable};
+
+	fn queue() -> BanningTransactionQueue {
+		BanningTransactionQueue::new(TransactionQueue::default(), Threshold::BanAfter(1), Duration::from_secs(180))
+	}
+
+	fn default_account_details(_address: &Address) -> AccountDetails {
+		AccountDetails {
+			nonce: U256::zero(),
+			balance: !U256::zero(),
+		}
+	}
+
+	fn transaction(action: Action) -> SignedTransaction {
+		let keypair = Random.generate().unwrap();
+		Transaction {
+			action: action,
+			value: U256::from(100),
+			data: "3331600055".from_hex().unwrap(),
+			gas: U256::from(100_000),
+			gas_price: U256::from(10),
+			nonce: U256::from(0),
+		}.sign(keypair.secret())
+	}
+
+	fn unwrap_err(res: Result<TransactionImportResult, Error>) -> TransactionError {
+		match res {
+			Err(Error::Transaction(e)) => e,
+			Ok(x) => panic!("Expected error, got: Ok({:?})", x),
+			Err(e) => panic!("Unexpected error type returned by queue: {:?}", e),
+		}
+	}
+
+	#[test]
+	fn should_allow_to_borrow_the_queue() {
+		// given
+		let tx = transaction(Action::Create);
+		let mut txq = queue();
+
+		// when
+		txq.queue().add(tx, &default_account_details, TransactionOrigin::External).unwrap();
+
+		// then
+		// should also deref to queue
+		assert_eq!(txq.status().pending, 1);
+	}
+
+	#[test]
+	fn should_not_accept_transactions_from_banned_sender() {
+		// given
+		let tx = transaction(Action::Create);
+		let mut txq = queue();
+		// Banlist once (threshold not reached)
+		let banlist1 = txq.ban_sender(tx.sender().unwrap());
+		assert!(!banlist1, "Threshold not reached yet.");
+		// Insert once
+		let import1 = txq.add_with_banlist(tx.clone(), &default_account_details).unwrap();
+		assert_eq!(import1, TransactionImportResult::Current);
+
+		// when
+		let banlist2 = txq.ban_sender(tx.sender().unwrap());
+		let import2 = txq.add_with_banlist(tx.clone(), &default_account_details);
+
+		// then
+		assert!(banlist2, "Threshold should be reached - banned.");
+		assert_eq!(unwrap_err(import2), TransactionError::SenderBanned);
+		// Should also remove transacion from the queue
+		assert_eq!(txq.find(&tx.hash()), None);
+	}
+
+	#[test]
+	fn should_not_accept_transactions_to_banned_recipient() {
+		// given
+		let recipient = Address::default();
+		let tx = transaction(Action::Call(recipient));
+		let mut txq = queue();
+		// Banlist once (threshold not reached)
+		let banlist1 = txq.ban_recipient(recipient);
+		assert!(!banlist1, "Threshold not reached yet.");
+		// Insert once
+		let import1 = txq.add_with_banlist(tx.clone(), &default_account_details).unwrap();
+		assert_eq!(import1, TransactionImportResult::Current);
+
+		// when
+		let banlist2 = txq.ban_recipient(recipient);
+		let import2 = txq.add_with_banlist(tx.clone(), &default_account_details);
+
+		// then
+		assert!(banlist2, "Threshold should be reached - banned.");
+		assert_eq!(unwrap_err(import2), TransactionError::RecipientBanned);
+	}
+
+	#[test]
+	fn should_not_accept_transactions_with_banned_code() {
+		// given
+		let tx = transaction(Action::Create);
+		let codehash = tx.data.sha3();
+		let mut txq = queue();
+		// Banlist once (threshold not reached)
+		let banlist1 = txq.ban_codehash(codehash);
+		assert!(!banlist1, "Threshold not reached yet.");
+		// Insert once
+		let import1 = txq.add_with_banlist(tx.clone(), &default_account_details).unwrap();
+		assert_eq!(import1, TransactionImportResult::Current);
+
+		// when
+		let banlist2 = txq.ban_codehash(codehash);
+		let import2 = txq.add_with_banlist(tx.clone(), &default_account_details);
+
+		// then
+		assert!(banlist2, "Threshold should be reached - banned.");
+		assert_eq!(unwrap_err(import2), TransactionError::CodeBanned);
+	}
+}

ethcore/src/miner/miner.rs

@@ -31,6 +31,7 @@ use receipt::{Receipt, RichReceipt};
 use spec::Spec;
 use engines::Engine;
 use miner::{MinerService, MinerStatus, TransactionQueue, PrioritizationStrategy, AccountDetails, TransactionOrigin};
+use miner::banning_queue::{BanningTransactionQueue, Threshold};
 use miner::work_notify::WorkPoster;
 use client::TransactionImportResult;
 use miner::price_info::PriceInfo;
@@ -59,6 +60,22 @@ pub enum GasLimit {
 	Fixed(U256),
 }
 
+/// Transaction queue banning settings.
+#[derive(Debug, PartialEq, Clone)]
+pub enum Banning {
+	/// Banning in transaction queue is disabled
+	Disabled,
+	/// Banning in transaction queue is enabled
+	Enabled {
+		/// Upper limit of transaction processing time before banning.
+		offend_threshold: Duration,
+		/// Number of similar offending transactions before banning.
+		min_offends: u16,
+		/// Number of seconds the offender is banned for.
+		ban_duration: Duration,
+	},
+}
+
 /// Configures the behaviour of the miner.
 #[derive(Debug, PartialEq)]
 pub struct MinerOptions {
@@ -86,6 +103,8 @@ pub struct MinerOptions {
 	pub enable_resubmission: bool,
 	/// Global gas limit for all transaction in the queue except for local and retracted.
 	pub tx_queue_gas_limit: GasLimit,
+	/// Banning settings
+	pub tx_queue_banning: Banning,
 }
 
 impl Default for MinerOptions {
@@ -103,6 +122,7 @@ impl Default for MinerOptions {
 			reseal_min_period: Duration::from_secs(2),
 			work_queue_size: 20,
 			enable_resubmission: true,
+			tx_queue_banning: Banning::Disabled,
 		}
 	}
 }
@@ -186,7 +206,7 @@ struct SealingWork {
 /// Handles preparing work for "work sealing" or seals "internally" if Engine does not require work.
 pub struct Miner {
 	// NOTE [ToDr]  When locking always lock in this order!
-	transaction_queue: Arc<Mutex<TransactionQueue>>,
+	transaction_queue: Arc<Mutex<BanningTransactionQueue>>,
 	sealing_work: Mutex<SealingWork>,
 	next_allowed_reseal: Mutex<Instant>,
 	sealing_block_last_request: Mutex<u64>,
@@ -215,11 +235,18 @@ impl Miner {
 			GasLimit::Fixed(ref limit) => *limit,
 			_ => !U256::zero(),
 		};
-		let txq = Arc::new(Mutex::new(TransactionQueue::with_limits(
-			options.tx_queue_strategy, options.tx_queue_size, gas_limit, options.tx_gas_limit
-		)));
+
+		let txq = TransactionQueue::with_limits(options.tx_queue_strategy, options.tx_queue_size, gas_limit, options.tx_gas_limit);
+		let txq = match options.tx_queue_banning {
+			Banning::Disabled => BanningTransactionQueue::new(txq, Threshold::NeverBan, Duration::from_secs(180)),
+			Banning::Enabled { ban_duration, min_offends, .. } => BanningTransactionQueue::new(
+				txq,
+				Threshold::BanAfter(min_offends),
+				ban_duration,
+			),
+		};
 		Miner {
-			transaction_queue: txq,
+			transaction_queue: Arc::new(Mutex::new(txq)),
 			next_allowed_reseal: Mutex::new(Instant::now()),
 			sealing_block_last_request: Mutex::new(0),
 			sealing_work: Mutex::new(SealingWork{
@@ -318,10 +345,31 @@ impl Miner {
 		let mut invalid_transactions = HashSet::new();
 		let mut transactions_to_penalize = HashSet::new();
 		let block_number = open_block.block().fields().header.number();
-		// TODO: push new uncles, too.
+
+		// TODO Push new uncles too.
 		for tx in transactions {
 			let hash = tx.hash();
-			match open_block.push_transaction(tx, None) {
+			let start = Instant::now();
+			let result = open_block.push_transaction(tx, None);
+			let took = start.elapsed();
+
+			// Check for heavy transactions
+			match self.options.tx_queue_banning {
+				Banning::Enabled { ref offend_threshold, .. } if &took > offend_threshold => {
+					match self.transaction_queue.lock().ban_transaction(&hash) {
+						true => {
+							warn!(target: "miner", "Detected heavy transaction. Banning the sender and recipient/code.");
+						},
+						false => {
+							transactions_to_penalize.insert(hash);
+							debug!(target: "miner", "Detected heavy transaction. Penalizing sender.")
+						}
+					}
+				},
+				_ => {},
+			}
+
+			match result {
 				Err(Error::Execution(ExecutionError::BlockGasLimitReached { gas_limit, gas_used, gas })) => {
 					debug!(target: "miner", "Skipping adding transaction to block because of gas limit: {:?} (limit: {:?}, used: {:?}, gas: {:?})", hash, gas_limit, gas_used, gas);
 
@@ -506,7 +554,7 @@ impl Miner {
 		prepare_new
 	}
 
-	fn add_transactions_to_queue(&self, chain: &MiningBlockChainClient, transactions: Vec<SignedTransaction>, origin: TransactionOrigin, transaction_queue: &mut TransactionQueue) ->
+	fn add_transactions_to_queue(&self, chain: &MiningBlockChainClient, transactions: Vec<SignedTransaction>, origin: TransactionOrigin, transaction_queue: &mut BanningTransactionQueue) ->
 		Vec<Result<TransactionImportResult, Error>> {
 
 		let fetch_account = |a: &Address| AccountDetails {
@@ -515,7 +563,14 @@ impl Miner {
 		};
 
 		transactions.into_iter()
-			.map(|tx| transaction_queue.add(tx, &fetch_account, origin))
+			.map(|tx| match origin {
+				TransactionOrigin::Local | TransactionOrigin::RetractedBlock => {
+					transaction_queue.add(tx, &fetch_account, origin)
+				},
+				TransactionOrigin::External => {
+					transaction_queue.add_with_banlist(tx, &fetch_account)
+				}
+			})
 			.collect()
 	}
 
@@ -1099,6 +1154,7 @@ mod tests {
 				pending_set: PendingSet::AlwaysSealing,
 				work_queue_size: 5,
 				enable_resubmission: true,
+				tx_queue_banning: Banning::Disabled,
 			},
 			GasPricer::new_fixed(0u64.into()),
 			&Spec::new_test(),

ethcore/src/miner/mod.rs

@@ -44,11 +44,12 @@
 mod miner;
 mod external;
 mod transaction_queue;
+mod banning_queue;
 mod work_notify;
 mod price_info;
 
 pub use self::transaction_queue::{TransactionQueue, PrioritizationStrategy, AccountDetails, TransactionOrigin};
-pub use self::miner::{Miner, MinerOptions, PendingSet, GasPricer, GasPriceCalibratorOptions, GasLimit};
+pub use self::miner::{Miner, MinerOptions, Banning, PendingSet, GasPricer, GasPriceCalibratorOptions, GasLimit};
 pub use self::external::{ExternalMiner, ExternalMinerService};
 pub use client::TransactionImportResult;