Networking fixes

util/network/src/connection.rs

@@ -354,7 +354,10 @@ impl EncryptedConnection {
 	/// Send a packet
 	pub fn send_packet<Message>(&mut self, io: &IoContext<Message>, payload: &[u8]) -> Result<(), NetworkError> where Message: Send + Clone + Sync + 'static {
 		let mut header = RlpStream::new();
-		let len = payload.len() as usize;
+		let len = payload.len();
+		if len >= (1 << 24) {
+			return Err(NetworkError::OversizedPacket);
+		}
 		header.append_raw(&[(len >> 16) as u8, (len >> 8) as u8, len as u8], 1);
 		header.append_raw(&[0xc2u8, 0x80u8, 0x80u8], 1);
 		//TODO: ger rid of vectors here

util/network/src/error.rs

@@ -106,6 +106,8 @@ pub enum NetworkError {
 	AddressResolve(Option<::std::io::Error>),
 	/// Error concerning the Rust standard library's IO subsystem.
 	StdIo(::std::io::Error),
+	/// Packet size is over the protocol limit.
+	OversizedPacket,
 }
 
 impl fmt::Display for NetworkError {
@@ -124,6 +126,7 @@ impl fmt::Display for NetworkError {
 			AddressResolve(_) => "Failed to resolve network address.".into(),
 			StdIo(ref err) => format!("{}", err),
 			Util(ref err) => format!("{}", err),
+			OversizedPacket => "Packet is too large".into(),
 		};
 
 		f.write_fmt(format_args!("Network error ({})", msg))

util/network/src/host.rs

@@ -844,7 +844,8 @@ impl Host {
 								// only proceed if the connecting peer is reserved.
 								if !self.reserved_nodes.read().contains(&id) {
 									s.disconnect(io, DisconnectReason::TooManyPeers);
-									return;
+									kill = true;
+									break;
 								}
 							}
 							ready_id = Some(id);
@@ -895,6 +896,7 @@ impl Host {
 				if duplicate {
 					trace!(target: "network", "Rejected duplicate connection: {}", token);
 					session.lock().disconnect(io, DisconnectReason::DuplicatePeer);
+					self.kill_connection(token, io, false);
 					return;
 				}
 				for p in ready_data {
@@ -1159,8 +1161,11 @@ impl IoHandler<NetworkIoMessage> for Host {
 			FIRST_SESSION ... LAST_SESSION => {
 				let mut connections = self.sessions.write();
 				if let Some(connection) = connections.get(stream).cloned() {
-					connection.lock().deregister_socket(event_loop).expect("Error deregistering socket");
-					connections.remove(stream);
+					let c = connection.lock();
+					if c.expired() { // make sure it is the same connection that the event was generated for
+						c.deregister_socket(event_loop).expect("Error deregistering socket");
+						connections.remove(stream);
+					}
 				}
 			}
 			DISCOVERY => (),

util/network/src/session.rs

@@ -241,7 +241,7 @@ impl Session {
 	/// Check if this session is expired.
 	pub fn expired(&self) -> bool {
 		match self.state {
-			State::Handshake(ref h) => h.expired(),
+			State::Handshake(ref h) => self.expired || h.expired(),
 			_ => self.expired,
 		}
 	}
@@ -407,7 +407,7 @@ impl Session {
 				let rlp = UntrustedRlp::new(&packet.data[1..]);
 				let reason: u8 = rlp.val_at(0)?;
 				if self.had_hello {
-					debug!("Disconnected: {}: {:?}", self.token(), DisconnectReason::from_u8(reason));
+					debug!(target:"network", "Disconnected: {}: {:?}", self.token(), DisconnectReason::from_u8(reason));
 				}
 				Err(From::from(NetworkError::Disconnect(DisconnectReason::from_u8(reason))))
 			}