Simplify and improve headers_received (#1362)

p2p/src/msg.rs

@@ -604,9 +604,13 @@ impl Readable for Headers {
 		if (len as u32) > MAX_BLOCK_HEADERS + 1 {
 			return Err(ser::Error::TooLargeReadErr);
 		}
-		let mut headers = Vec::with_capacity(len as usize);
-		for _ in 0..len {
-			headers.push(BlockHeader::read(reader)?);
+		let mut headers: Vec<BlockHeader> = Vec::with_capacity(len as usize);
+		for n in 0..len as usize {
+			let header = BlockHeader::read(reader)?;
+			if n > 0 && header.height != headers[n - 1].height + 1 {
+				return Err(ser::Error::CorruptedData);
+			}
+			headers.push(header);
 		}
 		Ok(Headers { headers: headers })
 	}

p2p/src/peer.rs

@@ -23,8 +23,9 @@ use core::core::target::Difficulty;
 use handshake::Handshake;
 use msg::{self, BanReason, GetPeerAddrs, Locator, Ping, TxHashSetRequest};
 use protocol::Protocol;
-use types::{Capabilities, ChainAdapter, Error, NetAdapter, P2PConfig, PeerInfo, ReasonForBan,
-            TxHashSetRead};
+use types::{
+	Capabilities, ChainAdapter, Error, NetAdapter, P2PConfig, PeerInfo, ReasonForBan, TxHashSetRead,
+};
 use util::LOGGER;
 
 const MAX_TRACK_SIZE: usize = 30;
@@ -160,7 +161,8 @@ impl Peer {
 	/// Send the ban reason before banning
 	pub fn send_ban_reason(&self, ban_reason: ReasonForBan) {
 		let ban_reason_msg = BanReason { ban_reason };
-		match self.connection
+		match self
+			.connection
 			.as_ref()
 			.unwrap()
 			.send(ban_reason_msg, msg::Type::BanReason)
@@ -412,7 +414,7 @@ impl ChainAdapter for TrackingAdapter {
 		self.adapter.header_received(bh, addr)
 	}
 
-	fn headers_received(&self, bh: Vec<core::BlockHeader>, addr: SocketAddr) {
+	fn headers_received(&self, bh: Vec<core::BlockHeader>, addr: SocketAddr) -> bool {
 		self.adapter.headers_received(bh, addr)
 	}
 
@@ -432,17 +434,8 @@ impl ChainAdapter for TrackingAdapter {
 		self.adapter.txhashset_receive_ready()
 	}
 
-	fn txhashset_write(
-		&self,
-		h: Hash,
-		txhashset_data: File,
-		peer_addr: SocketAddr,
-	) -> bool {
-		self.adapter.txhashset_write(
-			h,
-			txhashset_data,
-			peer_addr,
-		)
+	fn txhashset_write(&self, h: Hash, txhashset_data: File, peer_addr: SocketAddr) -> bool {
+		self.adapter.txhashset_write(h, txhashset_data, peer_addr)
 	}
 }
 

p2p/src/peers.rs

@@ -19,10 +19,10 @@ use std::sync::{Arc, RwLock};
 
 use rand::{thread_rng, Rng};
 
+use chrono::prelude::Utc;
 use core::core;
 use core::core::hash::{Hash, Hashed};
 use core::core::target::Difficulty;
-use chrono::prelude::{Utc};
 use util::LOGGER;
 
 use peer::Peer;
@@ -111,7 +111,8 @@ impl Peers {
 
 	/// Get vec of peers we are currently connected to.
 	pub fn connected_peers(&self) -> Vec<Arc<RwLock<Peer>>> {
-		let mut res = self.peers
+		let mut res = self
+			.peers
 			.read()
 			.unwrap()
 			.values()
@@ -263,9 +264,7 @@ impl Peers {
 			error!(LOGGER, "Couldn't ban {}: {:?}", peer_addr, e);
 		}
 
-		if let Err(e) =
-			self.update_last_banned(peer_addr.clone(), Utc::now().timestamp())
-		{
+		if let Err(e) = self.update_last_banned(peer_addr.clone(), Utc::now().timestamp()) {
 			error!(
 				LOGGER,
 				"Couldn't update last_banned time {}: {:?}", peer_addr, e
@@ -597,8 +596,15 @@ impl ChainAdapter for Peers {
 		}
 	}
 
-	fn headers_received(&self, headers: Vec<core::BlockHeader>, peer_addr: SocketAddr) {
-		self.adapter.headers_received(headers, peer_addr)
+	fn headers_received(&self, headers: Vec<core::BlockHeader>, peer_addr: SocketAddr) -> bool {
+		if !self.adapter.headers_received(headers, peer_addr) {
+			// if the peer sent us a block header that's intrinsically bad
+			// they are either mistaken or malevolent, both of which require a ban
+			self.ban_peer(&peer_addr, ReasonForBan::BadBlockHeader);
+			false
+		} else {
+			true
+		}
 	}
 
 	fn locate_headers(&self, hs: Vec<Hash>) -> Vec<core::BlockHeader> {

p2p/src/serv.rs

@@ -239,7 +239,9 @@ impl ChainAdapter for DummyAdapter {
 	fn block_received(&self, _: core::Block, _: SocketAddr) -> bool {
 		true
 	}
-	fn headers_received(&self, _: Vec<core::BlockHeader>, _: SocketAddr) {}
+	fn headers_received(&self, _: Vec<core::BlockHeader>, _: SocketAddr) -> bool {
+		true
+	}
 	fn locate_headers(&self, _: Vec<Hash>) -> Vec<core::BlockHeader> {
 		vec![]
 	}

p2p/src/types.rs

@@ -242,7 +242,7 @@ pub trait ChainAdapter: Sync + Send {
 	/// A set of block header has been received, typically in response to a
 	/// block
 	/// header request.
-	fn headers_received(&self, bh: Vec<core::BlockHeader>, addr: SocketAddr);
+	fn headers_received(&self, bh: Vec<core::BlockHeader>, addr: SocketAddr) -> bool;
 
 	/// Finds a list of block headers based on the provided locator. Tries to
 	/// identify the common chain and gets the headers that follow it

servers/src/common/adapters.rs

@@ -228,7 +228,7 @@ impl p2p::ChainAdapter for NetToChainAdapter {
 		true
 	}
 
-	fn headers_received(&self, bhs: Vec<core::BlockHeader>, addr: SocketAddr) {
+	fn headers_received(&self, bhs: Vec<core::BlockHeader>, addr: SocketAddr) -> bool {
 		info!(
 			LOGGER,
 			"Received block headers {:?} from {}",
@@ -236,52 +236,34 @@ impl p2p::ChainAdapter for NetToChainAdapter {
 			addr,
 		);
 
+		if bhs.len() == 0 {
+			return false;
+		}
+
+		// headers will just set us backward if even the last is unknown
+		let last_h = bhs.last().unwrap().hash();
+		if let Ok(_) = w(&self.chain).get_block_header(&last_h) {
+			info!(LOGGER, "All known, ignoring");
+			return true;
+		}
+
 		// try to add each header to our header chain
-		let mut added_hs = vec![];
 		for bh in bhs {
 			let res = w(&self.chain).sync_block_header(&bh, self.chain_opts());
-			match res {
-				Ok(_) => {
-					added_hs.push(bh.hash());
-				}
-				Err(e) => {
-					match e.kind() {
-						chain::ErrorKind::Unfit(s) => {
-							info!(
-								LOGGER,
-								"Received unfit block header {} at {}: {}.",
-								bh.hash(),
-								bh.height,
-								s
-							);
-						}
-						chain::ErrorKind::StoreErr(e, explanation) => {
-							error!(
-								LOGGER,
-								"Store error processing block header {}: in {} {:?}",
-								bh.hash(),
-								explanation,
-								e
-							);
-							return;
-						}
-						_ => {
-							info!(LOGGER, "Invalid block header {}: {:?}.", bh.hash(), e);
-							// TODO penalize peer somehow
-						}
-					}
+			if let &Err(ref e) = &res {
+				debug!(
+					LOGGER,
+					"Block header {} refused by chain: {:?}",
+					bh.hash(),
+					e
+				);
+
+				if e.is_bad_data() {
+					return false;
 				}
 			}
 		}
-
-		let header_head = w(&self.chain).get_header_head().unwrap();
-		info!(
-			LOGGER,
-			"Added {} headers to the header chain. Last: {} at {}.",
-			added_hs.len(),
-			header_head.last_block_h,
-			header_head.height,
-		);
+		true
 	}
 
 	fn locate_headers(&self, locator: Vec<Hash>) -> Vec<core::BlockHeader> {