[WIP] P2P: initial I2P support

api/src/handlers/peers_api.rs

@@ -65,15 +65,15 @@ impl Handler for PeerHandler {
 		if let Ok(ip_addr) = command.parse() {
 			peer_addr = PeerAddr::from_ip(ip_addr);
 		} else if let Ok(addr) = command.parse() {
-			peer_addr = PeerAddr(addr);
+			peer_addr = PeerAddr::Socket(addr);
 		} else {
 			return response(
 				StatusCode::BAD_REQUEST,
 				format!("peer address unrecognized: {}", req.uri().path()),
 			);
 		}
 
-		match w_fut!(&self.peers).get_peer(peer_addr) {
+		match w_fut!(&self.peers).get_peer(&peer_addr) {
 			Ok(peer) => json_response(&peer),
 			Err(_) => response(StatusCode::NOT_FOUND, "peer not found"),
 		}
@@ -90,7 +90,7 @@ impl Handler for PeerHandler {
 				if let Ok(ip_addr) = a.parse() {
 					PeerAddr::from_ip(ip_addr)
 				} else if let Ok(addr) = a.parse() {
-					PeerAddr(addr)
+					PeerAddr::Socket(addr)
 				} else {
 					return response(
 						StatusCode::BAD_REQUEST,
@@ -101,8 +101,8 @@ impl Handler for PeerHandler {
 		};
 
 		match command {
-			"ban" => w_fut!(&self.peers).ban_peer(addr, ReasonForBan::ManualBan),
-			"unban" => w_fut!(&self.peers).unban_peer(addr),
+			"ban" => w_fut!(&self.peers).ban_peer(&addr, ReasonForBan::ManualBan),
+			"unban" => w_fut!(&self.peers).unban_peer(&addr),
 			_ => return response(StatusCode::BAD_REQUEST, "invalid command"),
 		};
 

chain/Cargo.toml

@@ -20,7 +20,7 @@ serde = "1"
 serde_derive = "1"
 chrono = "0.4.4"
 lru-cache = "0.1"
-lazy_static = "1"
+lazy_static = "1.3.0"
 regex = "1"
 
 grin_core = { path = "../core", version = "2.0.1-beta.1" }

config/src/comments.rs

@@ -249,11 +249,40 @@ fn comments() -> HashMap<String, String> {
 		.to_string(),
 	);
 
+	retval.insert(
+		"i2p_mode".to_string(),
+		"
+#Configuration for I2P support (\"Disabled\" by default)
+#I2P can be enabled by the following:
+#[server.p2p_config.i2p_mode]
+#mode = \"Enabled\"
+
+#[server.p2p_config.i2p_mode.i2p_config]
+#autostart = true/false  - start I2P automatically
+#exclusive = true/false  - exclusively connect through I2P, or also use TCP
+#addr = \"127.0.0.1:7656\"  - address of local I2P server
+"
+		.to_string(),
+	);
+
 	retval.insert(
 		"[server.p2p_config.capabilities]".to_string(),
 		"#If the seeding type is List, the list of peers to connect to can
-#be specified as follows:
-#seeds = [\"192.168.0.1:3414\",\"192.168.0.2:3414\"]
+#be specified as follows (same for other peer types):
+#For TCP addresses:
+#[[server.p2p_config.seeds]]
+#peer_type = \"Socket\"
+#peer_addr = \"192.168.0.1:3414\"
+
+#For I2P addresses:
+#[[server.p2p_config.seeds]]
+#peer_type = \"I2p\"
+
+#[server.p2p_config.seeds.peer_addr]
+#port = 3414
+
+#[server.p2p_config.seeds.peer_addr.dest]
+#inner = \"4jcuryak7cabdqrv72ytqonf4xnk3jcryfkzqz43wo67dqwlh2xa.b32.i2p\"
 
 #hardcoded peer lists for allow/deny
 #will *only* connect to peers in allow list

p2p/Cargo.toml

@@ -13,6 +13,7 @@ edition = "2018"
 bitflags = "1"
 bytes = "0.4"
 enum_primitive = "0.1"
+i2p = { version = "0.1.0", git = "https://github.com/ignopeverell/i2p-rs"}
 net2 = "0.2"
 num = "0.1"
 rand = "0.6"
@@ -29,3 +30,4 @@ grin_chain = { path = "../chain", version = "2.0.1-beta.1" }
 
 [dev-dependencies]
 grin_pool = { path = "../pool", version = "2.0.1-beta.1" }
+toml = "0.4"

p2p/src/conn.rs

@@ -22,7 +22,7 @@
 
 use std::fs::File;
 use std::io::{self, Read, Write};
-use std::net::{Shutdown, TcpStream};
+use std::net::Shutdown;
 use std::sync::{mpsc, Arc};
 use std::{
 	cmp,
@@ -35,7 +35,7 @@ use crate::msg::{
 	read_body, read_discard, read_header, read_item, write_to_buf, MsgHeader, MsgHeaderWrapper,
 	Type,
 };
-use crate::types::Error;
+use crate::types::{Error, Stream};
 use crate::util::read_write::{read_exact, write_all};
 use crate::util::{RateCounter, RwLock};
 
@@ -262,7 +262,7 @@ impl Tracker {
 /// the current thread, instead just returns a future and the Connection
 /// itself.
 pub fn listen<H>(
-	stream: TcpStream,
+	stream: Stream,
 	version: ProtocolVersion,
 	tracker: Arc<Tracker>,
 	handler: H,
@@ -290,7 +290,7 @@ where
 }
 
 fn poll<H>(
-	conn: TcpStream,
+	conn: Stream,
 	version: ProtocolVersion,
 	handler: H,
 	send_rx: mpsc::Receiver<Vec<u8>>,

p2p/src/handshake.rs

@@ -17,11 +17,13 @@ use crate::core::pow::Difficulty;
 use crate::core::ser::ProtocolVersion;
 use crate::msg::{read_message, write_message, Hand, Shake, Type, USER_AGENT};
 use crate::peer::Peer;
-use crate::types::{Capabilities, Direction, Error, P2PConfig, PeerAddr, PeerInfo, PeerLiveInfo};
+use crate::types::{
+	Capabilities, Direction, Error, P2PConfig, PeerAddr, PeerInfo, PeerLiveInfo, Stream,
+};
 use crate::util::RwLock;
 use rand::{thread_rng, Rng};
 use std::collections::VecDeque;
-use std::net::{SocketAddr, TcpStream};
+use std::net::SocketAddr;
 use std::sync::Arc;
 
 /// Local generated nonce for peer connecting.
@@ -64,14 +66,11 @@ impl Handshake {
 		capabilities: Capabilities,
 		total_difficulty: Difficulty,
 		self_addr: PeerAddr,
-		conn: &mut TcpStream,
+		conn: &mut Stream,
 	) -> Result<PeerInfo, Error> {
 		// prepare the first part of the handshake
 		let nonce = self.next_nonce();
-		let peer_addr = match conn.peer_addr() {
-			Ok(pa) => PeerAddr(pa),
-			Err(e) => return Err(Error::Connection(e)),
-		};
+		let peer_addr = conn.peer_addr()?;
 
 		// Using our default version here.
 		let version = ProtocolVersion::default();
@@ -83,7 +82,7 @@ impl Handshake {
 			genesis: self.genesis,
 			total_difficulty,
 			sender_addr: self_addr,
-			receiver_addr: peer_addr,
+			receiver_addr: peer_addr.clone(),
 			user_agent: USER_AGENT.to_string(),
 		};
 
@@ -102,15 +101,15 @@ impl Handshake {
 		let peer_info = PeerInfo {
 			capabilities: shake.capabilities,
 			user_agent: shake.user_agent,
-			addr: peer_addr,
+			addr: peer_addr.clone(),
 			version: shake.version,
 			live_info: Arc::new(RwLock::new(PeerLiveInfo::new(shake.total_difficulty))),
 			direction: Direction::Outbound,
 		};
 
 		// If denied then we want to close the connection
 		// (without providing our peer with any details why).
-		if Peer::is_denied(&self.config, peer_info.addr) {
+		if Peer::is_denied(&self.config, &peer_info.addr.clone()) {
 			return Err(Error::ConnectionClose);
 		}
 
@@ -129,7 +128,7 @@ impl Handshake {
 		&self,
 		capab: Capabilities,
 		total_difficulty: Difficulty,
-		conn: &mut TcpStream,
+		conn: &mut Stream,
 	) -> Result<PeerInfo, Error> {
 		// Note: We read the Hand message *before* we know which protocol version
 		// is supported by our peer (it is in the Hand message).
@@ -145,7 +144,7 @@ impl Handshake {
 		} else {
 			// check the nonce to see if we are trying to connect to ourselves
 			let nonces = self.nonces.read();
-			let addr = resolve_peer_addr(hand.sender_addr, &conn);
+			let addr = resolve_peer_addr(hand.sender_addr.clone(), &conn);
 			if nonces.contains(&hand.nonce) {
 				// save ip addresses of ourselves
 				let mut addrs = self.addrs.write();
@@ -171,7 +170,7 @@ impl Handshake {
 		// so check if we are configured to explicitly allow or deny it.
 		// If denied then we want to close the connection
 		// (without providing our peer with any details why).
-		if Peer::is_denied(&self.config, peer_info.addr) {
+		if Peer::is_denied(&self.config, &peer_info.addr.clone()) {
 			return Err(Error::ConnectionClose);
 		}
 
@@ -205,11 +204,20 @@ impl Handshake {
 }
 
 /// Resolve the correct peer_addr based on the connection and the advertised port.
-fn resolve_peer_addr(advertised: PeerAddr, conn: &TcpStream) -> PeerAddr {
-	let port = advertised.0.port();
-	if let Ok(addr) = conn.peer_addr() {
-		PeerAddr(SocketAddr::new(addr.ip(), port))
-	} else {
-		advertised
+fn resolve_peer_addr(advertised: PeerAddr, conn: &Stream) -> PeerAddr {
+	match advertised.clone() {
+		PeerAddr::Socket(addr) => {
+			let port = addr.port();
+			if let Ok(addr) = conn.peer_addr() {
+				if let Ok(ip) = addr.unwrap_ip() {
+					PeerAddr::Socket(SocketAddr::new(ip.ip(), port))
+				} else {
+					advertised
+				}
+			} else {
+				advertised
+			}
+		}
+		PeerAddr::I2p(_) => advertised,
 	}
 }

p2p/src/lib.rs

@@ -52,6 +52,6 @@ pub use crate::peers::Peers;
 pub use crate::serv::{DummyAdapter, Server};
 pub use crate::store::{PeerData, State};
 pub use crate::types::{
-	Capabilities, ChainAdapter, Direction, Error, P2PConfig, PeerAddr, PeerInfo, ReasonForBan,
-	Seeding, TxHashSetRead, MAX_BLOCK_HEADERS, MAX_LOCATORS, MAX_PEER_ADDRS,
+	Capabilities, ChainAdapter, Direction, Error, I2pMode, P2PConfig, PeerAddr, PeerInfo,
+	ReasonForBan, Seeding, TxHashSetRead, MAX_BLOCK_HEADERS, MAX_LOCATORS, MAX_PEER_ADDRS,
 };

p2p/src/peer.rs

@@ -16,7 +16,7 @@ use crate::util::{Mutex, RwLock};
 use std::fmt;
 use std::fs::File;
 use std::io::Read;
-use std::net::{Shutdown, TcpStream};
+use std::net::Shutdown;
 use std::path::PathBuf;
 use std::sync::Arc;
 
@@ -33,7 +33,7 @@ use crate::msg::{
 use crate::protocol::Protocol;
 use crate::types::{
 	Capabilities, ChainAdapter, Error, NetAdapter, P2PConfig, PeerAddr, PeerInfo, ReasonForBan,
-	TxHashSetRead,
+	Stream, TxHashSetRead,
 };
 use chrono::prelude::{DateTime, Utc};
 
@@ -70,7 +70,7 @@ impl fmt::Debug for Peer {
 
 impl Peer {
 	// Only accept and connect can be externally used to build a peer
-	fn new(info: PeerInfo, conn: TcpStream, adapter: Arc<dyn NetAdapter>) -> std::io::Result<Peer> {
+	fn new(info: PeerInfo, conn: Stream, adapter: Arc<dyn NetAdapter>) -> std::io::Result<Peer> {
 		let state = Arc::new(RwLock::new(State::Connected));
 		let tracking_adapter = TrackingAdapter::new(adapter);
 		let handler = Protocol::new(Arc::new(tracking_adapter.clone()), info.clone());
@@ -89,7 +89,7 @@ impl Peer {
 	}
 
 	pub fn accept(
-		mut conn: TcpStream,
+		mut conn: Stream,
 		capab: Capabilities,
 		total_difficulty: Difficulty,
 		hs: &Handshake,
@@ -114,7 +114,7 @@ impl Peer {
 	}
 
 	pub fn connect(
-		mut conn: TcpStream,
+		mut conn: Stream,
 		capab: Capabilities,
 		total_difficulty: Difficulty,
 		self_addr: PeerAddr,
@@ -139,9 +139,25 @@ impl Peer {
 		}
 	}
 
-	pub fn is_denied(config: &P2PConfig, peer_addr: PeerAddr) -> bool {
+	/// Main peer loop listening for messages and forwarding to the rest of the
+	/// system.
+	pub fn start(&mut self, conn: Stream) -> Result<(), Error> {
+		let adapter = Arc::new(self.tracking_adapter.clone());
+		let handler = Protocol::new(adapter, self.info.clone());
+		let (sendh, stoph) = conn::listen(
+			conn,
+			self.info.version.clone(),
+			self.tracker.clone(),
+			handler,
+		)?;
+		self.send_handle = Mutex::new(sendh);
+		self.stop_handle = Mutex::new(stoph);
+		Ok(())
+	}
+
+	pub fn is_denied(config: &P2PConfig, peer_addr: &PeerAddr) -> bool {
 		if let Some(ref denied) = config.peers_deny {
-			if denied.contains(&peer_addr) {
+			if denied.contains(peer_addr) {
 				debug!(
 					"checking peer allowed/denied: {:?} explicitly denied",
 					peer_addr
@@ -150,7 +166,7 @@ impl Peer {
 			}
 		}
 		if let Some(ref allowed) = config.peers_allow {
-			if allowed.contains(&peer_addr) {
+			if allowed.contains(peer_addr) {
 				debug!(
 					"checking peer allowed/denied: {:?} explicitly allowed",
 					peer_addr
@@ -603,11 +619,11 @@ impl NetAdapter for TrackingAdapter {
 		self.adapter.peer_addrs_received(addrs)
 	}
 
-	fn peer_difficulty(&self, addr: PeerAddr, diff: Difficulty, height: u64) {
+	fn peer_difficulty(&self, addr: &PeerAddr, diff: Difficulty, height: u64) {
 		self.adapter.peer_difficulty(addr, diff, height)
 	}
 
-	fn is_banned(&self, addr: PeerAddr) -> bool {
+	fn is_banned(&self, addr: &PeerAddr) -> bool {
 		self.adapter.is_banned(addr)
 	}
 }