[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

@@ -257,11 +257,43 @@ 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]
+# Start I2P automatically
+#autostart = true
+# Exclusively connect through I2P, or also use TCP/IP
+#exclusive = true
+# Address of local I2P server
+#addr = \"127.0.0.1:7656\"
+"
+		.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/mimblewimble/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

@@ -26,11 +26,11 @@ 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::{RateCounter, RwLock};
 use std::fs::File;
 use std::io::{self, Read, Write};
-use std::net::{Shutdown, TcpStream};
+use std::net::Shutdown;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{mpsc, Arc};
 use std::time::Duration;
@@ -273,23 +273,20 @@ impl Tracker {
 /// Start listening on the provided connection and wraps it. Does not hang
 /// the current thread, instead just returns a future and the Connection
 /// itself.
-pub fn listen<H>(
-	stream: TcpStream,
+pub fn listen<S, H>(
+	stream: S,
 	version: ProtocolVersion,
 	tracker: Arc<Tracker>,
 	handler: H,
-) -> io::Result<(ConnHandle, StopHandle)>
+) -> Result<(ConnHandle, StopHandle), Error>
 where
+	S: Stream + 'static,
 	H: MessageHandler,
 {
 	let (send_tx, send_rx) = mpsc::sync_channel(SEND_CHANNEL_CAP);
 
-	stream
-		.set_read_timeout(Some(IO_TIMEOUT))
-		.expect("can't set read timeout");
-	stream
-		.set_write_timeout(Some(IO_TIMEOUT))
-		.expect("can't set read timeout");
+	stream.set_read_timeout(Some(IO_TIMEOUT))?;
+	stream.set_write_timeout(Some(IO_TIMEOUT))?;
 
 	let stopped = Arc::new(AtomicBool::new(false));
 
@@ -308,15 +305,16 @@ where
 	))
 }
 
-fn poll<H>(
-	conn: TcpStream,
+fn poll<S, H>(
+	conn: S,
 	version: ProtocolVersion,
 	handler: H,
 	send_rx: mpsc::Receiver<Vec<u8>>,
 	stopped: Arc<AtomicBool>,
 	tracker: Arc<Tracker>,
 ) -> io::Result<(JoinHandle<()>, JoinHandle<()>)>
 where
+	S: Stream + 'static,
 	H: MessageHandler,
 {
 	// Split out tcp stream out into separate reader/writer halves.

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.
@@ -59,19 +61,16 @@ impl Handshake {
 		}
 	}
 
-	pub fn initiate(
+	pub fn initiate<S: Stream>(
 		&self,
 		capabilities: Capabilities,
 		total_difficulty: Difficulty,
 		self_addr: PeerAddr,
-		conn: &mut TcpStream,
+		conn: &mut S,
 	) -> 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 "local" protocol version.
 		let version = ProtocolVersion::local();
@@ -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);
 		}
 
@@ -125,17 +124,17 @@ impl Handshake {
 		Ok(peer_info)
 	}
 
-	pub fn accept(
+	pub fn accept<S: Stream>(
 		&self,
 		capab: Capabilities,
 		total_difficulty: Difficulty,
-		conn: &mut TcpStream,
+		mut conn: S,
 	) -> Result<PeerInfo, Error> {
 		// Note: We read the Hand message *before* we know which protocol version
 		// is supported by our peer (in the Hand message).
 		let version = ProtocolVersion::local();
 
-		let hand: Hand = read_message(conn, version, Type::Hand)?;
+		let hand: Hand = read_message(&mut conn, version, Type::Hand)?;
 
 		// all the reasons we could refuse this connection for
 		if hand.genesis != self.genesis {
@@ -146,7 +145,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(), &mut conn);
 			if nonces.contains(&hand.nonce) {
 				// save ip addresses of ourselves
 				let mut addrs = self.addrs.write();
@@ -172,7 +171,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) {
 			return Err(Error::ConnectionClose);
 		}
 
@@ -185,7 +184,7 @@ impl Handshake {
 			user_agent: USER_AGENT.to_string(),
 		};
 
-		write_message(conn, shake, Type::Shake, version)?;
+		write_message(&mut conn, shake, Type::Shake, version)?;
 		trace!("Success handshake with {}.", peer_info.addr);
 
 		Ok(peer_info)
@@ -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<S: Stream>(advertised: PeerAddr, conn: &S) -> PeerAddr {
+	match advertised {
+		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,
 };