Connection filter (#6359)

Cargo.lock

@@ -1615,6 +1615,19 @@ dependencies = [
  "ws2_32-sys 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "node-filter"
+version = "1.8.0"
+dependencies = [
+ "ethcore 1.8.0",
+ "ethcore-io 1.8.0",
+ "ethcore-network 1.8.0",
+ "ethcore-util 1.8.0",
+ "futures 0.1.11 (registry+https://github.com/rust-lang/crates.io-index)",
+ "log 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",
+ "native-contracts 0.1.0",
+]
+
 [[package]]
 name = "node-health"
 version = "0.1.0"
@@ -1838,6 +1851,7 @@ dependencies = [
  "isatty 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "jsonrpc-core 7.0.0 (git+https://github.com/paritytech/jsonrpc.git?branch=parity-1.7)",
  "log 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",
+ "node-filter 1.8.0",
  "node-health 0.1.0",
  "num_cpus 1.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "number_prefix 0.2.5 (registry+https://github.com/rust-lang/crates.io-index)",

Cargo.toml

@@ -42,6 +42,7 @@ ethcore-light = { path = "ethcore/light" }
 ethcore-logger = { path = "logger" }
 ethcore-stratum = { path = "stratum" }
 ethcore-network = { path = "util/network" }
+node-filter = { path = "ethcore/node_filter" }
 ethkey = { path = "ethkey" }
 node-health = { path = "dapps/node-health" }
 rlp = { path = "util/rlp" }

ethcore/native_contracts/build.rs

@@ -28,6 +28,7 @@ const SERVICE_TRANSACTION_ABI: &'static str = include_str!("res/service_transact
 const SECRETSTORE_ACL_STORAGE_ABI: &'static str = include_str!("res/secretstore_acl_storage.json");
 const VALIDATOR_SET_ABI: &'static str = include_str!("res/validator_set.json");
 const VALIDATOR_REPORT_ABI: &'static str = include_str!("res/validator_report.json");
+const PEER_SET_ABI: &'static str = include_str!("res/peer_set.json");
 
 const TEST_VALIDATOR_SET_ABI: &'static str = include_str!("res/test_validator_set.json");
 
@@ -53,6 +54,7 @@ fn main() {
 	build_file("SecretStoreAclStorage", SECRETSTORE_ACL_STORAGE_ABI, "secretstore_acl_storage.rs");
 	build_file("ValidatorSet", VALIDATOR_SET_ABI, "validator_set.rs");
 	build_file("ValidatorReport", VALIDATOR_REPORT_ABI, "validator_report.rs");
+	build_file("PeerSet", PEER_SET_ABI, "peer_set.rs");
 
 	build_test_contracts();
 }

ethcore/native_contracts/res/peer_set.json

@@ -0,0 +1 @@
+[{"constant":true,"inputs":[{"name":"sl","type":"bytes32"},{"name":"sh","type":"bytes32"},{"name":"pl","type":"bytes32"},{"name":"ph","type":"bytes32"}],"name":"connectionAllowed","outputs":[{"name":"res","type":"bool"}],"payable":false,"type":"function"},{"inputs":[],"payable":false,"type":"constructor"}]

ethcore/native_contracts/src/lib.rs

@@ -30,6 +30,7 @@ mod service_transaction;
 mod secretstore_acl_storage;
 mod validator_set;
 mod validator_report;
+mod peer_set;
 
 pub mod test_contracts;
 
@@ -40,3 +41,4 @@ pub use self::service_transaction::ServiceTransactionChecker;
 pub use self::secretstore_acl_storage::SecretStoreAclStorage;
 pub use self::validator_set::ValidatorSet;
 pub use self::validator_report::ValidatorReport;
+pub use self::peer_set::PeerSet;

ethcore/native_contracts/src/peer_set.rs

@@ -0,0 +1,21 @@
+// Copyright 2015-2017 Parity Technologies (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/>.
+
+#![allow(unused_mut, unused_variables, unused_imports)]
+
+//! Peer set contract.
+
+include!(concat!(env!("OUT_DIR"), "/peer_set.rs"));

ethcore/node_filter/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+description = "Parity smart network connections"
+homepage = "http://parity.io"
+license = "GPL-3.0"
+name = "node-filter"
+version = "1.8.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+
+[dependencies]
+ethcore = { path = ".."}
+ethcore-util = { path = "../../util" }
+ethcore-io = { path = "../../util/io" }
+ethcore-network = { path = "../../util/network" }
+native-contracts = { path = "../native_contracts" }
+futures = "0.1"
+log = "0.3"

ethcore/node_filter/res/node_filter.json

@@ -0,0 +1,44 @@
+{
+	"name": "TestNodeFilterContract",
+	"engine": {
+		"authorityRound": {
+			"params": {
+				"stepDuration": 1,
+				"startStep": 2,
+				"validators": {
+					"contract": "0x0000000000000000000000000000000000000005"
+				}
+			}
+		}
+	},
+	"params": {
+		"accountStartNonce": "0x0",
+		"maximumExtraDataSize": "0x20",
+		"minGasLimit": "0x1388",
+		"networkID" : "0x69",
+		"gasLimitBoundDivisor": "0x0400"
+	},
+	"genesis": {
+		"seal": {
+			"generic": "0xc180"
+		},
+		"difficulty": "0x20000",
+		"author": "0x0000000000000000000000000000000000000000",
+		"timestamp": "0x00",
+		"parentHash": "0x0000000000000000000000000000000000000000000000000000000000000000",
+		"extraData": "0x",
+		"gasLimit": "0x222222"
+	},
+	"accounts": {
+		"0000000000000000000000000000000000000001": { "balance": "1", "builtin": { "name": "ecrecover", "pricing": { "linear": { "base": 3000, "word": 0 } } } },
+		"0000000000000000000000000000000000000002": { "balance": "1", "builtin": { "name": "sha256", "pricing": { "linear": { "base": 60, "word": 12 } } } },
+		"0000000000000000000000000000000000000003": { "balance": "1", "builtin": { "name": "ripemd160", "pricing": { "linear": { "base": 600, "word": 120 } } } },
+		"0000000000000000000000000000000000000004": { "balance": "1", "builtin": { "name": "identity", "pricing": { "linear": { "base": 15, "word": 3 } } } },
+		"0000000000000000000000000000000000000005": {
+			"balance": "1",
+			"constructor": "6060604052341561000f57600080fd5b5b6012600102600080601160010260001916815260200190815260200160002081600019169055506022600102600080602160010260001916815260200190815260200160002081600019169055506032600102600080603160010260001916815260200190815260200160002081600019169055506042600102600080604160010260001916815260200190815260200160002081600019169055505b5b610155806100bd6000396000f30060606040526000357c0100000000000000000000000000000000000000000000000000000000900463ffffffff168063994d790a1461003e575b600080fd5b341561004957600080fd5b61008a6004808035600019169060200190919080356000191690602001909190803560001916906020019091908035600019169060200190919050506100a4565b604051808215151515815260200191505060405180910390f35b60006001800285600019161480156100c3575060026001028460001916145b156100d15760019050610121565b60006001028360001916141580156100f157506000600102826000191614155b801561011e5750816000191660008085600019166000191681526020019081526020016000205460001916145b90505b9493505050505600a165627a7a723058202082b8d8667fd397925f39785d8e804540beda0524d28af15921375145dfcc250029"
+		},
+		"0x7d577a597b2742b498cb5cf0c26cdcd726d39e6e": { "balance": "1606938044258990275541962092341162602522202993782792835301376" },
+		"0x82a978b3f5962a5b0957d9ee9eef472ee55b42f1": { "balance": "1606938044258990275541962092341162602522202993782792835301376" }
+	}
+}

ethcore/node_filter/src/lib.rs

@@ -0,0 +1,154 @@
+// Copyright 2015-2017 Parity Technologies (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/>.
+
+//! Smart contract based node filter.
+
+extern crate ethcore;
+extern crate ethcore_util as util;
+extern crate ethcore_network as network;
+extern crate native_contracts;
+extern crate futures;
+#[cfg(test)] extern crate ethcore_io as io;
+#[macro_use] extern crate log;
+
+use std::sync::Weak;
+use std::collections::HashMap;
+use native_contracts::PeerSet as Contract;
+use network::{NodeId, ConnectionFilter, ConnectionDirection};
+use ethcore::client::{BlockChainClient, BlockId, ChainNotify};
+use util::{Mutex, Address, H256, Bytes};
+use futures::Future;
+
+const MAX_CACHE_SIZE: usize = 4096;
+
+/// Connection filter that uses a contract to manage permissions.
+pub struct NodeFilter {
+	contract: Mutex<Option<Contract>>,
+	client: Weak<BlockChainClient>,
+	contract_address: Address,
+	permission_cache: Mutex<HashMap<NodeId, bool>>,
+}
+
+impl NodeFilter {
+	/// Create a new instance. Accepts a contract address.
+	pub fn new(client: Weak<BlockChainClient>, contract_address: Address) -> NodeFilter {
+		NodeFilter {
+			contract: Mutex::new(None),
+			client: client,
+			contract_address: contract_address,
+			permission_cache: Mutex::new(HashMap::new()),
+		}
+	}
+
+	/// Clear cached permissions.
+	pub fn clear_cache(&self) {
+		self.permission_cache.lock().clear();
+	}
+}
+
+impl ConnectionFilter for NodeFilter {
+	fn connection_allowed(&self, own_id: &NodeId, connecting_id: &NodeId, _direction: ConnectionDirection) -> bool {
+
+		let mut cache = self.permission_cache.lock();
+		if let Some(res) = cache.get(connecting_id) {
+			return *res;
+		}
+
+		let mut contract = self.contract.lock();
+		if contract.is_none() {
+			*contract = Some(Contract::new(self.contract_address));
+		}
+
+		let allowed = match (self.client.upgrade(), &*contract) {
+			(Some(ref client), &Some(ref contract)) => {
+				let own_low = H256::from_slice(&own_id[0..32]);
+				let own_high = H256::from_slice(&own_id[32..64]);
+				let id_low = H256::from_slice(&connecting_id[0..32]);
+				let id_high = H256::from_slice(&connecting_id[32..64]);
+				let allowed = contract.connection_allowed(
+					|addr, data| futures::done(client.call_contract(BlockId::Latest, addr, data)),
+					own_low,
+					own_high,
+					id_low,
+					id_high,
+				).wait().unwrap_or_else(|e| {
+					debug!("Error callling peer set contract: {:?}", e);
+					false
+				});
+
+				allowed
+			}
+			_ => false,
+		};
+
+		if cache.len() < MAX_CACHE_SIZE {
+			cache.insert(*connecting_id, allowed);
+		}
+		allowed
+	}
+}
+
+impl ChainNotify for NodeFilter {
+	fn new_blocks(&self, imported: Vec<H256>, _invalid: Vec<H256>, _enacted: Vec<H256>, _retracted: Vec<H256>, _sealed: Vec<H256>, _proposed: Vec<Bytes>, _duration: u64) {
+		if !imported.is_empty() {
+			self.clear_cache();
+		}
+	}
+}
+
+
+#[cfg(test)]
+mod test {
+	use std::sync::{Arc, Weak};
+	use std::str::FromStr;
+	use ethcore::spec::Spec;
+	use ethcore::client::{BlockChainClient, Client, ClientConfig};
+	use ethcore::miner::Miner;
+	use util::{Address};
+	use network::{ConnectionDirection, ConnectionFilter, NodeId};
+	use io::IoChannel;
+	use super::NodeFilter;
+
+	/// Contract code: https://gist.github.com/arkpar/467dbcc73cbb85b0997a7a10ffa0695f
+	#[test]
+	fn node_filter() {
+		let contract_addr = Address::from_str("0000000000000000000000000000000000000005").unwrap();
+		let data = include_bytes!("../res/node_filter.json");
+		let spec = Spec::load(::std::env::temp_dir(), &data[..]).unwrap();
+		let client_db = Arc::new(::util::kvdb::in_memory(::ethcore::db::NUM_COLUMNS.unwrap_or(0)));
+
+		let client = Client::new(
+			ClientConfig::default(),
+			&spec,
+			client_db,
+			Arc::new(Miner::with_spec(&spec)),
+			IoChannel::disconnected(),
+		).unwrap();
+		let filter = NodeFilter::new(Arc::downgrade(&client) as Weak<BlockChainClient>, contract_addr);
+		let self1 = NodeId::from_str("00000000000000000000000000000000000000000000000000000000000000010000000000000000000000000000000000000000000000000000000000000002").unwrap();
+		let self2 = NodeId::from_str("00000000000000000000000000000000000000000000000000000000000000020000000000000000000000000000000000000000000000000000000000000003").unwrap();
+		let node1 = NodeId::from_str("00000000000000000000000000000000000000000000000000000000000000110000000000000000000000000000000000000000000000000000000000000012").unwrap();
+		let node2 = NodeId::from_str("00000000000000000000000000000000000000000000000000000000000000210000000000000000000000000000000000000000000000000000000000000022").unwrap();
+		let nodex = NodeId::from_str("77000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000").unwrap();
+
+		assert!(filter.connection_allowed(&self1, &node1, ConnectionDirection::Inbound));
+		assert!(filter.connection_allowed(&self1, &nodex, ConnectionDirection::Inbound));
+		filter.clear_cache();
+		assert!(filter.connection_allowed(&self2, &node1, ConnectionDirection::Inbound));
+		assert!(filter.connection_allowed(&self2, &node2, ConnectionDirection::Inbound));
+		assert!(!filter.connection_allowed(&self2, &nodex, ConnectionDirection::Inbound));
+	}
+}

ethcore/src/spec/spec.rs

@@ -100,6 +100,8 @@ pub struct CommonParams {
 	pub block_reward: U256,
 	/// Registrar contract address.
 	pub registrar: Address,
+	/// Node permission managing contract address.
+	pub node_permission_contract: Option<Address>,
 }
 
 impl CommonParams {
@@ -171,6 +173,7 @@ impl From<ethjson::spec::Params> for CommonParams {
 			gas_limit_bound_divisor: p.gas_limit_bound_divisor.into(),
 			block_reward: p.block_reward.map_or_else(U256::zero, Into::into),
 			registrar: p.registrar.map_or_else(Address::new, Into::into),
+			node_permission_contract: p.node_permission_contract.map(Into::into),
 		}
 	}
 }

json/src/spec/params.rs

@@ -102,6 +102,9 @@ pub struct Params {
 	pub block_reward: Option<Uint>,
 	/// See `CommonParams` docs.
 	pub registrar: Option<Address>,
+	/// Node permission contract address.
+	#[serde(rename="nodePermissionContract")]
+	pub node_permission_contract: Option<Address>,
 }
 
 #[cfg(test)]

parity/main.rs

@@ -69,6 +69,7 @@ extern crate parity_updater as updater;
 extern crate parity_whisper;
 extern crate path;
 extern crate rpc_cli;
+extern crate node_filter;
 
 #[macro_use]
 extern crate log as rlog;

parity/modules.rs

@@ -19,7 +19,7 @@ use std::path::Path;
 
 use ethcore::client::BlockChainClient;
 use hypervisor::Hypervisor;
-use ethsync::{AttachedProtocol, SyncConfig, NetworkConfiguration, NetworkError, Params};
+use ethsync::{AttachedProtocol, SyncConfig, NetworkConfiguration, NetworkError, Params, ConnectionFilter};
 use ethcore::snapshot::SnapshotService;
 use light::Provider;
 
@@ -183,6 +183,7 @@ pub fn sync(
 	provider: Arc<Provider>,
 	_log_settings: &LogConfig,
 	attached_protos: Vec<AttachedProtocol>,
+	connection_filter: Option<Arc<ConnectionFilter>>,
 ) -> Result<SyncModules, NetworkError> {
 	let eth_sync = EthSync::new(Params {
 		config: sync_cfg,
@@ -191,7 +192,8 @@ pub fn sync(
 		snapshot_service: snapshot_service,
 		network_config: net_cfg,
 		attached_protos: attached_protos,
-	})?;
+	},
+	connection_filter)?;
 
 	Ok((eth_sync.clone() as Arc<SyncProvider>, eth_sync.clone() as Arc<ManageNetwork>, eth_sync.clone() as Arc<ChainNotify>))
 }

parity/run.rs

@@ -15,7 +15,7 @@
 // along with Parity.  If not, see <http://www.gnu.org/licenses/>.
 
 use std::fmt;
-use std::sync::Arc;
+use std::sync::{Arc, Weak};
 use std::net::{TcpListener};
 
 use ctrlc::CtrlC;
@@ -38,6 +38,7 @@ use parity_reactor::EventLoop;
 use parity_rpc::{NetworkSettings, informant, is_major_importing};
 use updater::{UpdatePolicy, Updater};
 use util::{Colour, version, Mutex, Condvar};
+use node_filter::NodeFilter;
 
 use params::{
 	SpecType, Pruning, AccountsConfig, GasPricerConfig, MinerExtras, Switch,
@@ -569,11 +570,13 @@ pub fn execute(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) -> R
 		miner.clone(),
 	).map_err(|e| format!("Client service error: {:?}", e))?;
 
+	let connection_filter_address = spec.params().node_permission_contract;
 	// drop the spec to free up genesis state.
 	drop(spec);
 
 	// take handle to client
 	let client = service.client();
+	let connection_filter = connection_filter_address.map(|a| Arc::new(NodeFilter::new(Arc::downgrade(&client) as Weak<BlockChainClient>, a)));
 	let snapshot_service = service.snapshot_service();
 
 	// initialize the local node information store.
@@ -645,9 +648,13 @@ pub fn execute(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) -> R
 		client.clone(),
 		&cmd.logger_config,
 		attached_protos,
+		connection_filter.clone().map(|f| f as Arc<::ethsync::ConnectionFilter + 'static>),
 	).map_err(|e| format!("Sync error: {}", e))?;
 
 	service.add_notify(chain_notify.clone());
+	if let Some(filter) = connection_filter {
+		service.add_notify(filter);
+	}
 
 	// start network
 	if network_enabled {

sync/src/api.rs

@@ -19,7 +19,7 @@ use std::collections::{HashMap, BTreeMap};
 use std::io;
 use util::Bytes;
 use network::{NetworkProtocolHandler, NetworkService, NetworkContext, HostInfo, PeerId, ProtocolId,
-	NetworkConfiguration as BasicNetworkConfiguration, NonReservedPeerMode, NetworkError};
+	NetworkConfiguration as BasicNetworkConfiguration, NonReservedPeerMode, NetworkError, ConnectionFilter};
 use util::{U256, H256, H512};
 use io::{TimerToken};
 use ethcore::ethstore::ethkey::Secret;
@@ -236,7 +236,7 @@ pub struct EthSync {
 
 impl EthSync {
 	/// Creates and register protocol with the network service
-	pub fn new(params: Params) -> Result<Arc<EthSync>, NetworkError> {
+	pub fn new(params: Params, connection_filter: Option<Arc<ConnectionFilter>>) -> Result<Arc<EthSync>, NetworkError> {
 		const MAX_LIGHTSERV_LOAD: f64 = 0.5;
 
 		let pruning_info = params.chain.pruning_info();
@@ -272,7 +272,7 @@ impl EthSync {
 		};
 
 		let chain_sync = ChainSync::new(params.config, &*params.chain);
-		let service = NetworkService::new(params.network_config.clone().into_basic()?)?;
+		let service = NetworkService::new(params.network_config.clone().into_basic()?, connection_filter)?;
 
 		let sync = Arc::new(EthSync {
 			network: service,
@@ -736,7 +736,7 @@ impl LightSync {
 			(sync_handler, Arc::new(light_proto))
 		};
 
-		let service = NetworkService::new(params.network_config)?;
+		let service = NetworkService::new(params.network_config, None)?;
 
 		Ok(LightSync {
 			proto: light_proto,

sync/src/lib.rs

@@ -76,7 +76,7 @@ mod api;
 
 pub use api::*;
 pub use chain::{SyncStatus, SyncState};
-pub use network::{is_valid_node_url, NonReservedPeerMode, NetworkError};
+pub use network::{is_valid_node_url, NonReservedPeerMode, NetworkError, ConnectionFilter, ConnectionDirection};
 
 /// IPC interfaces
 #[cfg(feature="ipc")]

util/network/src/connection_filter.rs

@@ -0,0 +1,31 @@
+// Copyright 2015-2017 Parity Technologies (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/>.
+
+//! Connection filter trait.
+
+use super::NodeId;
+
+/// Filtered connection direction.
+pub enum ConnectionDirection {
+	Inbound,
+	Outbound,
+}
+
+/// Connection filter. Each connection is checked against `connection_allowed`.
+pub trait ConnectionFilter : Send + Sync {
+	/// Filter a connection. Returns `true` if connection should be allowed. `false` if rejected.
+	fn connection_allowed(&self, own_id: &NodeId, connecting_id: &NodeId, direction: ConnectionDirection) -> bool;
+}

util/network/src/host.rs

@@ -42,6 +42,7 @@ use discovery::{Discovery, TableUpdates, NodeEntry};
 use ip_utils::{map_external_address, select_public_address};
 use path::restrict_permissions_owner;
 use parking_lot::{Mutex, RwLock};
+use connection_filter::{ConnectionFilter, ConnectionDirection};
 
 type Slab<T> = ::slab::Slab<T, usize>;
 
@@ -380,11 +381,12 @@ pub struct Host {
 	reserved_nodes: RwLock<HashSet<NodeId>>,
 	num_sessions: AtomicUsize,
 	stopping: AtomicBool,
+	filter: Option<Arc<ConnectionFilter>>,
 }
 
 impl Host {
 	/// Create a new instance
-	pub fn new(mut config: NetworkConfiguration, stats: Arc<NetworkStats>) -> Result<Host, NetworkError> {
+	pub fn new(mut config: NetworkConfiguration, stats: Arc<NetworkStats>, filter: Option<Arc<ConnectionFilter>>) -> Result<Host, NetworkError> {
 		let mut listen_address = match config.listen_address {
 			None => SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), DEFAULT_PORT)),
 			Some(addr) => addr,
@@ -437,6 +439,7 @@ impl Host {
 			reserved_nodes: RwLock::new(HashSet::new()),
 			num_sessions: AtomicUsize::new(0),
 			stopping: AtomicBool::new(false),
+			filter: filter,
 		};
 
 		for n in boot_nodes {
@@ -691,8 +694,12 @@ impl Host {
 
 		let max_handshakes_per_round = max_handshakes / 2;
 		let mut started: usize = 0;
-		for id in nodes.filter(|id| !self.have_session(id) && !self.connecting_to(id) && *id != self_id)
+		for id in nodes.filter(|id| 
-			.take(min(max_handshakes_per_round, max_handshakes - handshake_count)) {
+				!self.have_session(id) &&
+				!self.connecting_to(id) &&
+				*id != self_id &&
+				self.filter.as_ref().map_or(true, |f| f.connection_allowed(&self_id, &id, ConnectionDirection::Outbound))
+			).take(min(max_handshakes_per_round, max_handshakes - handshake_count)) {
 			self.connect_peer(&id, io);
 			started += 1;
 		}
@@ -827,7 +834,7 @@ impl Host {
 						Ok(SessionData::Ready) => {
 							self.num_sessions.fetch_add(1, AtomicOrdering::SeqCst);
 							let session_count = self.session_count();
-							let (min_peers, max_peers, reserved_only) = {
+							let (min_peers, max_peers, reserved_only, self_id) = {
 								let info = self.info.read();
 								let mut max_peers = info.config.max_peers;
 								for cap in s.info.capabilities.iter() {
@@ -836,7 +843,7 @@ impl Host {
 										break;
 									}
 								}
-								(info.config.min_peers as usize, max_peers as usize, info.config.non_reserved_mode == NonReservedPeerMode::Deny)
+								(info.config.min_peers as usize, max_peers as usize, info.config.non_reserved_mode == NonReservedPeerMode::Deny, info.id().clone())
 							};
 
 							let id = s.id().expect("Ready session always has id").clone();
@@ -852,6 +859,14 @@ impl Host {
 									break;
 								}
 							}
+
+							if !self.filter.as_ref().map_or(true, |f| f.connection_allowed(&self_id, &id, ConnectionDirection::Inbound)) {
+								trace!(target: "network", "Inbound connection not allowed for {:?}", id);
+								s.disconnect(io, DisconnectReason::UnexpectedIdentity);
+								kill = true;
+								break;
+							}
+
 							ready_id = Some(id);
 
 							// Add it to the node table
@@ -1266,7 +1281,7 @@ fn host_client_url() {
 	let mut config = NetworkConfiguration::new_local();
 	let key = "6f7b0d801bc7b5ce7bbd930b84fd0369b3eb25d09be58d64ba811091046f3aa2".parse().unwrap();
 	config.use_secret = Some(key);
-	let host: Host = Host::new(config, Arc::new(NetworkStats::new())).unwrap();
+	let host: Host = Host::new(config, Arc::new(NetworkStats::new()), None).unwrap();
 	assert!(host.local_url().starts_with("enode://101b3ef5a4ea7a1c7928e24c4c75fd053c235d7b80c22ae5c03d145d0ac7396e2a4ffff9adee3133a7b05044a5cee08115fd65145e5165d646bde371010d803c@"));
 }
 

util/network/src/lib.rs

@@ -44,7 +44,7 @@
 //! }
 //!
 //! fn main () {
-//! 	let mut service = NetworkService::new(NetworkConfiguration::new_local()).expect("Error creating network service");
+//! 	let mut service = NetworkService::new(NetworkConfiguration::new_local(), None).expect("Error creating network service");
 //! 	service.start().expect("Error starting service");
 //! 	service.register_protocol(Arc::new(MyHandler), *b"myp", 1, &[1u8]);
 //!
@@ -95,6 +95,7 @@ mod error;
 mod node_table;
 mod stats;
 mod ip_utils;
+mod connection_filter;
 
 #[cfg(test)]
 mod tests;
@@ -104,6 +105,7 @@ pub use service::NetworkService;
 pub use error::NetworkError;
 pub use stats::NetworkStats;
 pub use session::SessionInfo;
+pub use connection_filter::{ConnectionFilter, ConnectionDirection};
 
 pub use io::TimerToken;
 pub use node_table::{is_valid_node_url, NodeId};

util/network/src/service.rs

@@ -22,6 +22,7 @@ use io::*;
 use parking_lot::RwLock;
 use std::sync::Arc;
 use ansi_term::Colour;
+use connection_filter::ConnectionFilter;
 
 struct HostHandler {
 	public_url: RwLock<Option<String>>
@@ -48,11 +49,12 @@ pub struct NetworkService {
 	stats: Arc<NetworkStats>,
 	host_handler: Arc<HostHandler>,
 	config: NetworkConfiguration,
+	filter: Option<Arc<ConnectionFilter>>,
 }
 
 impl NetworkService {
 	/// Starts IO event loop
-	pub fn new(config: NetworkConfiguration) -> Result<NetworkService, NetworkError> {
+	pub fn new(config: NetworkConfiguration, filter: Option<Arc<ConnectionFilter>>) -> Result<NetworkService, NetworkError> {
 		let host_handler = Arc::new(HostHandler { public_url: RwLock::new(None) });
 		let io_service = IoService::<NetworkIoMessage>::start()?;
 
@@ -65,6 +67,7 @@ impl NetworkService {
 			host: RwLock::new(None),
 			config: config,
 			host_handler: host_handler,
+			filter: filter,
 		})
 	}
 
@@ -115,7 +118,7 @@ impl NetworkService {
 	pub fn start(&self) -> Result<(), NetworkError> {
 		let mut host = self.host.write();
 		if host.is_none() {
-			let h = Arc::new(Host::new(self.config.clone(), self.stats.clone())?);
+			let h = Arc::new(Host::new(self.config.clone(), self.stats.clone(), self.filter.clone())?);
 			self.io_service.register_handler(h.clone())?;
 			*host = Some(h);
 		}

util/network/src/tests.rs

@@ -92,7 +92,7 @@ impl NetworkProtocolHandler for TestProtocol {
 
 #[test]
 fn net_service() {
-	let service = NetworkService::new(NetworkConfiguration::new_local()).expect("Error creating network service");
+	let service = NetworkService::new(NetworkConfiguration::new_local(), None).expect("Error creating network service");
 	service.start().unwrap();
 	service.register_protocol(Arc::new(TestProtocol::new(false)), *b"myp", 1, &[1u8]).unwrap();
 }
@@ -104,13 +104,13 @@ fn net_connect() {
 	let mut config1 = NetworkConfiguration::new_local();
 	config1.use_secret = Some(key1.secret().clone());
 	config1.boot_nodes = vec![ ];
-	let mut service1 = NetworkService::new(config1).unwrap();
+	let mut service1 = NetworkService::new(config1, None).unwrap();
 	service1.start().unwrap();
 	let handler1 = TestProtocol::register(&mut service1, false);
 	let mut config2 = NetworkConfiguration::new_local();
 	info!("net_connect: local URL: {}", service1.local_url().unwrap());
 	config2.boot_nodes = vec![ service1.local_url().unwrap() ];
-	let mut service2 = NetworkService::new(config2).unwrap();
+	let mut service2 = NetworkService::new(config2, None).unwrap();
 	service2.start().unwrap();
 	let handler2 = TestProtocol::register(&mut service2, false);
 	while !handler1.got_packet() && !handler2.got_packet() && (service1.stats().sessions() == 0 || service2.stats().sessions() == 0) {
@@ -123,7 +123,7 @@ fn net_connect() {
 #[test]
 fn net_start_stop() {
 	let config = NetworkConfiguration::new_local();
-	let service = NetworkService::new(config).unwrap();
+	let service = NetworkService::new(config, None).unwrap();
 	service.start().unwrap();
 	service.stop().unwrap();
 	service.start().unwrap();
@@ -135,12 +135,12 @@ fn net_disconnect() {
 	let mut config1 = NetworkConfiguration::new_local();
 	config1.use_secret = Some(key1.secret().clone());
 	config1.boot_nodes = vec![ ];
-	let mut service1 = NetworkService::new(config1).unwrap();
+	let mut service1 = NetworkService::new(config1, None).unwrap();
 	service1.start().unwrap();
 	let handler1 = TestProtocol::register(&mut service1, false);
 	let mut config2 = NetworkConfiguration::new_local();
 	config2.boot_nodes = vec![ service1.local_url().unwrap() ];
-	let mut service2 = NetworkService::new(config2).unwrap();
+	let mut service2 = NetworkService::new(config2, None).unwrap();
 	service2.start().unwrap();
 	let handler2 = TestProtocol::register(&mut service2, true);
 	while !(handler1.got_disconnect() && handler2.got_disconnect()) {
@@ -153,7 +153,7 @@ fn net_disconnect() {
 #[test]
 fn net_timeout() {
 	let config = NetworkConfiguration::new_local();
-	let mut service = NetworkService::new(config).unwrap();
+	let mut service = NetworkService::new(config, None).unwrap();
 	service.start().unwrap();
 	let handler = TestProtocol::register(&mut service, false);
 	while !handler.got_timeout() {