Browse Source

The testnet can now create tunnels through IP protocol

interface-2
Lev 3 years ago
parent
commit
02c88de37a
  1. 205
      src/interfaces/ip.rs
  2. 226
      src/ironforce.rs
  3. 2
      src/res.rs
  4. 19
      src/tunnel.rs

205
src/interfaces/ip.rs

@ -3,16 +3,15 @@ use alloc::string::{String, ToString};
use alloc::vec; use alloc::vec;
use alloc::vec::Vec; use alloc::vec::Vec;
use core::ops::RangeInclusive; use core::ops::RangeInclusive;
#[cfg(test)]
use core::str::FromStr;
use core::time::Duration; use core::time::Duration;
use rayon::prelude::*; use rayon::prelude::*;
use std::net;
use std::net::TcpStream; use std::net::TcpStream;
#[cfg(test)] use std::{format, net};
use std::thread;
use crate::interface::{Interface, InterfaceRequirements, TargetingData}; use crate::interface::{Interface, InterfaceRequirements, TargetingData};
use crate::message::MessageBytes; use crate::message::MessageBytes;
use crate::res::IFError::General;
use crate::res::{IFError, IFResult}; use crate::res::{IFError, IFResult};
use crate::std::io::{Read, Write}; use crate::std::io::{Read, Write};
use crate::std::println; use crate::std::println;
@ -30,10 +29,10 @@ pub struct IPInterface {
connections: Vec<net::TcpStream>, connections: Vec<net::TcpStream>,
listener: net::TcpListener, listener: net::TcpListener,
peers: Vec<Peer>, peers: Vec<Peer>,
package_queue: Vec<IPPackage>, package_queue: Vec<(IPPackage, String /* from_peer */)>,
} }
#[derive(Debug)] #[derive(Debug, Clone)]
struct IPPackage { struct IPPackage {
version: u8, version: u8,
package_type: MessageType, package_type: MessageType,
@ -41,7 +40,7 @@ struct IPPackage {
message: MessageBytes, message: MessageBytes,
} }
#[derive(Debug)] #[derive(Debug, Copy, Clone)]
enum MessageType { enum MessageType {
Common, Common,
PeerRequest, PeerRequest,
@ -58,7 +57,7 @@ impl MessageType {
} }
} }
fn to_u8(&self) -> u8 { fn as_u8(&self) -> u8 {
match self { match self {
MessageType::Common => 0, MessageType::Common => 0,
MessageType::PeerRequest => 1, MessageType::PeerRequest => 1,
@ -71,28 +70,47 @@ impl InterfaceRequirements for IPInterface {}
impl Interface for IPInterface { impl Interface for IPInterface {
fn main_loop_iteration(&mut self) -> IFResult<()> { fn main_loop_iteration(&mut self) -> IFResult<()> {
println!("Mainloop {:?}", self.listener.local_addr()); if let Some(conn) = self.listener.incoming().next() {
println!("Incoming connections: {:?}", self.listener.incoming()); match conn {
Ok(stream) => {
match self.listener.accept() { stream.set_nonblocking(true)?;
Ok((stream, addr)) => { let addr = stream.peer_addr()?;
println!("New client: {:?}", addr); println!(
"({:?}): New client: {:?}",
addr,
self.listener.local_addr().unwrap()
);
if self.peers.iter().all(|(ip, _)| *ip != addr.ip()) { if self.peers.iter().all(|(ip, _)| *ip != addr.ip()) {
self.peers.push((addr.ip(), addr.port())); self.peers.push((addr.ip(), addr.port()));
} }
self.connections.push(stream) self.connections.push(stream)
} }
Err(_e) => return Err(IFError::General(String::from("No incoming connection"))), Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {}
Err(e) => return Err(IFError::from(e)),
}
} }
println!("Hello from mainloop");
let mut new_connections: Vec<TcpStream> = vec![]; let mut new_connections: Vec<TcpStream> = vec![];
for connection in &mut self.connections { for connection in &mut self.connections {
connection.set_nonblocking(true)?;
let mut header: [u8; 6] = [0, 0, 0, 0, 0, 0]; let mut header: [u8; 6] = [0, 0, 0, 0, 0, 0];
connection.read_exact(&mut header)?; match connection.read_exact(&mut header) {
Ok(_) => {}
Err(ref e)
if e.kind() == std::io::ErrorKind::WouldBlock
|| e.kind() == std::io::ErrorKind::UnexpectedEof =>
{
continue
}
Err(e) => {
println!("Error: {:?}", e);
continue;
}
};
let version = header[0]; let version = header[0];
let package_type = MessageType::from_u8(header[1])?; let package_type = MessageType::from_u8(header[1])?;
let size = bytes_to_size([header[2], header[3], header[4], header[5]]); let size = bytes_to_size([header[2], header[3], header[4], header[5]]);
connection.set_nonblocking(false)?;
connection.set_read_timeout(Some(std::time::Duration::from_millis(500)))?;
let mut message_take = connection.take(size as u64); let mut message_take = connection.take(size as u64);
let mut message: Vec<u8> = vec![]; let mut message: Vec<u8> = vec![];
@ -110,7 +128,7 @@ impl Interface for IPInterface {
connection, connection,
IPPackage { IPPackage {
version, version,
package_type, package_type: MessageType::PeersShared,
size: message.len() as u32, size: message.len() as u32,
message, message,
}, },
@ -123,7 +141,8 @@ impl Interface for IPInterface {
size, size,
message, message,
}; };
self.package_queue.push(package); self.package_queue
.push((package, format!("{:?}", connection.peer_addr()?)));
} }
MessageType::PeersShared => { MessageType::PeersShared => {
let peers: Vec<Peer> = serde_cbor::from_slice(message.as_slice())?; let peers: Vec<Peer> = serde_cbor::from_slice(message.as_slice())?;
@ -153,62 +172,57 @@ impl Interface for IPInterface {
&*self.id &*self.id
} }
fn send(&mut self, message: &[u8], interface_data: Option<TargetingData>) -> IFResult<()> { fn send(&mut self, message: &[u8], interface_data: Option<TargetingData>) -> IFResult<()> {
let addr: net::SocketAddr = match interface_data {
Some(ip_string) => ip_string.parse().expect("Unable to parse address"),
None => {
return Err(IFError::General(String::from(
"Not enough info to create connections",
)))
}
};
println!("Connecting to {:?}", addr);
let index = match self
.connections
.iter()
.position(|connection| connection.peer_addr().ok() == Some(addr))
{
None => {
let connection = match net::TcpStream::connect_timeout(&addr, Duration::new(1, 0)) {
Ok(connection) => connection,
Err(_) => return Err(General(String::from("Can't connect to this port"))),
};
self.connections.push(connection);
self.connections.len() - 1
}
Some(i) => i,
};
println!( println!(
"Sending message to {:?}", "({:?}): Connecting to {:?} to send a message",
self.connections[index].peer_addr().unwrap() self.listener.local_addr().unwrap(),
interface_data
); );
self.connections[index].set_write_timeout(Some(Duration::new(1, 0)))?; let package = IPPackage {
IPInterface::send_package(
&mut self.connections[index],
IPPackage {
version: 0, version: 0,
package_type: MessageType::Common, package_type: MessageType::Common,
size: message.len() as u32, size: message.len() as u32,
message: Vec::from(message), message: Vec::from(message),
}, };
)?;
match interface_data {
Some(ip_string) => {
let addr: net::SocketAddr = ip_string.parse().expect("Unable to parse address");
let index = self.obtain_connection(&(addr.ip(), addr.port()))?;
println!(
"({:?}): We have a connection to {:?}",
self.listener.local_addr().unwrap(),
self.connections[index].peer_addr().unwrap()
);
IPInterface::send_package(&mut self.connections[index], package)?;
}
None => {
for conn in &mut self.connections {
IPInterface::send_package(conn, package.clone())?;
}
}
};
println!("Sent message"); println!("Sent message");
Ok(()) Ok(())
} }
fn receive(&mut self) -> IFResult<Option<(MessageBytes, TargetingData)>> { fn receive(&mut self) -> IFResult<Option<(MessageBytes, TargetingData)>> {
if !self.package_queue.is_empty() {
println!(
"({:?}): New message from {}",
self.listener.local_addr().unwrap(),
self.package_queue.last().unwrap().1
);
}
match self.package_queue.pop() { match self.package_queue.pop() {
Some(ip_package) => Ok(Some((ip_package.message, "".to_string()))), Some((ip_package, data)) => Ok(Some((ip_package.message, data))),
None => Ok(None), None => Ok(None),
} }
} }
} }
impl IPInterface { impl IPInterface {
pub fn new(peers: Vec<Peer>) -> IFResult<Self> { pub fn new(port: u16, peers: Vec<Peer>) -> IFResult<Self> {
let listener = match create_tcp_listener() { let listener = match create_tcp_listener(port) {
Some(listener) => listener, Some(listener) => listener,
None => { None => {
return Err(IFError::General(String::from( return Err(IFError::General(String::from(
@ -250,7 +264,10 @@ impl IPInterface {
} }
fn send_package(stream: &mut net::TcpStream, package: IPPackage) -> IFResult<()> { fn send_package(stream: &mut net::TcpStream, package: IPPackage) -> IFResult<()> {
let mut header: Vec<u8> = vec![package.version, package.package_type.to_u8()]; stream.set_write_timeout(Some(std::time::Duration::from_millis(700)))?;
#[cfg(test)]
stream.set_nonblocking(false)?;
let mut header: Vec<u8> = vec![package.version, package.package_type.as_u8()];
for byte in size_to_bytes(package.size) { for byte in size_to_bytes(package.size) {
header.push(byte); header.push(byte);
} }
@ -280,22 +297,26 @@ impl IPInterface {
Ok(()) Ok(())
} }
fn obtain_connection(&mut self, addr: &Peer) -> IFResult<()> { fn obtain_connection(&mut self, addr: &Peer) -> IFResult<usize> {
if self fn compare_addrs(peer: &Peer, addr: net::SocketAddr) -> bool {
.connections addr.ip() == peer.0 && addr.port() == peer.1
.iter() }
.any(|con| con.peer_addr().is_ok() && con.peer_addr().unwrap().ip() == addr.0)
{ if let Some(pos) = self.connections.iter().position(|con| {
return Ok(()); con.peer_addr().is_ok() && compare_addrs(addr, con.peer_addr().unwrap())
}) {
return Ok(pos);
} }
if let Some(conn) = Self::new_connection(addr)? { if let Some(conn) = Self::new_connection(addr)? {
self.connections.push(conn) self.connections.push(conn);
Ok(self.connections.len() - 1)
} else {
Err(IFError::CouldNotConnect)
} }
Ok(())
} }
fn new_connection(addr: &Peer) -> IFResult<Option<TcpStream>> { fn new_connection(addr: &Peer) -> IFResult<Option<TcpStream>> {
for port in addr.1 - 5..addr.1 + 5 { for port in addr.1..addr.1 + 3 {
match net::TcpStream::connect_timeout( match net::TcpStream::connect_timeout(
&net::SocketAddr::new(addr.0, port as u16), &net::SocketAddr::new(addr.0, port as u16),
Duration::from_millis(500), Duration::from_millis(500),
@ -310,8 +331,8 @@ impl IPInterface {
} }
} }
fn create_tcp_listener() -> Option<net::TcpListener> { fn create_tcp_listener(port: u16) -> Option<net::TcpListener> {
for port in SOCKET_RANGE { for port in port..port + 5 {
match net::TcpListener::bind("127.0.0.1:".to_owned() + &port.to_string()) { match net::TcpListener::bind("127.0.0.1:".to_owned() + &port.to_string()) {
Ok(listener) => return Some(listener), Ok(listener) => return Some(listener),
Err(_e) => {} Err(_e) => {}
@ -349,21 +370,24 @@ fn bytes_to_size(arr: [u8; 4]) -> u32 {
#[test] #[test]
fn test_creating_connection() -> IFResult<()> { fn test_creating_connection() -> IFResult<()> {
let message = *b"Hello world from ironforest"; let message = *b"Hello world from ironforest";
let original_msg_copy = message;
let mut interface1 = IPInterface::new(vec![])?; let mut interface1 = IPInterface::new(50000, vec![])?;
let mut interface2 = IPInterface::new(vec![])?; let mut interface2 = IPInterface::new(50001, vec![])?;
let t2 = std::thread::spawn(move || {
for _ in 0..30 {
interface2.main_loop_iteration().unwrap();
std::thread::sleep(std::time::Duration::from_millis(150));
}
interface2
});
let t1 = std::thread::spawn(move || { let t1 = std::thread::spawn(move || {
interface1 interface1
.send(&message, Some(String::from("127.0.0.1:50001"))) .send(&message, Some(String::from("127.0.0.1:50001")))
.unwrap(); .unwrap();
interface1 interface1
}); });
thread::sleep(Duration::from_millis(10));
let t2 = std::thread::spawn(move || {
interface2.main_loop_iteration().unwrap();
interface2
});
let res1 = t1.join(); let res1 = t1.join();
match res1 { match res1 {
Ok(_res) => { Ok(_res) => {
@ -377,8 +401,14 @@ fn test_creating_connection() -> IFResult<()> {
println!("Thread Ok"); println!("Thread Ok");
match res.receive() { match res.receive() {
Ok(tmp) => match tmp { Ok(tmp) => match tmp {
Some((message, _metadata)) => println!("Received {:?}", message), Some((message, _metadata)) => {
None => println!("None"), println!("Received {:?}", message);
assert_eq!(message, original_msg_copy)
}
None => {
println!("None");
panic!();
}
}, },
Err(e) => println!("{:?}", e), Err(e) => println!("{:?}", e),
} }
@ -389,4 +419,17 @@ fn test_creating_connection() -> IFResult<()> {
} }
#[cfg(test)] #[cfg(test)]
pub mod test_ip_interface {} fn create_test_interfaces(n: usize) -> impl Iterator<Item = IPInterface> {
let ip_addr = std::net::IpAddr::from_str("127.0.0.1").unwrap();
(0..n).map(move |i| {
IPInterface::new(
(5000 + 5 * i) as u16,
// (0..n)
// .filter(|j| *j != i)
// .map(|j| (ip_addr, (5000 + 5 * j) as u16))
// .collect(),
vec![(ip_addr, (5000 + 5 * ((i + 5) % n)) as u16)],
)
.unwrap()
})
}

226
src/ironforce.rs

@ -3,8 +3,13 @@ use crate::message::{Message, MessageType, ServiceMessageType};
use crate::res::{IFError, IFResult}; use crate::res::{IFError, IFResult};
use crate::transport::Transport; use crate::transport::Transport;
use crate::tunnel::{Tunnel, TunnelPublic}; use crate::tunnel::{Tunnel, TunnelPublic};
use alloc::collections::BTreeMap;
use alloc::vec; use alloc::vec;
use alloc::vec::Vec; use alloc::vec::Vec;
#[cfg(feature = "std")]
use std::println;
const TUNNEL_MAX_REPEAT_COUNT: u32 = 3;
/// Main worker /// Main worker
pub struct IronForce { pub struct IronForce {
@ -23,11 +28,19 @@ pub struct IronForce {
/// Tunnels that has not been confirmed yet (no backward spread) /// Tunnels that has not been confirmed yet (no backward spread)
/// ///
/// `[(Tunnel, Optional target node, local peer ids)]` /// `[(Tunnel, Optional target node, local peer ids)]`
tunnels_pending: Vec<(TunnelPublic, Option<PublicKey> /* target node */, (u64, u64) /* local peer ids */)>, tunnels_pending: Vec<(
TunnelPublic,
Option<PublicKey>, /* target node */
(u64, u64), /* local peer ids */
)>,
/// True if this instance has background thread /// True if this instance has background thread
has_background_worker: bool, has_background_worker: bool,
/// Messages that were already processed (stored to avoid "echo chambers") /// Messages that were already processed (stored to avoid "echo chambers")
processed_messages: Vec<u64>, processed_messages: Vec<u64>,
/// Counters of how many times the tunnel has passed through this node on its forward movement (so that we don't do a shitposting)
///
/// Maps tunnel's first local_id to the number
tunnel_counters: BTreeMap<u64, u32>,
} }
impl IronForce { impl IronForce {
@ -42,12 +55,19 @@ impl IronForce {
tunnels_pending: vec![], tunnels_pending: vec![],
has_background_worker: false, has_background_worker: false,
processed_messages: vec![], processed_messages: vec![],
tunnel_counters: Default::default(),
} }
} }
/// Create a new tunnel to another node /// Create a new tunnel to another node
fn initialize_tunnel_creation(&mut self, destination: PublicKey) -> IFResult<()> { fn initialize_tunnel_creation(&mut self, destination: PublicKey) -> IFResult<()> {
let tunnel = TunnelPublic::new_singlecast(); let tunnel = TunnelPublic::new_singlecast();
#[cfg(std)]
println!(
"[{}] Tunnel for initialization: {:?}",
self.short_id(),
tunnel
);
self.tunnels_pending self.tunnels_pending
.push((tunnel.clone(), Some(destination.clone()), (0, 0))); .push((tunnel.clone(), Some(destination.clone()), (0, 0)));
let message = Message::build() let message = Message::build()
@ -77,7 +97,16 @@ impl IronForce {
mut message: Message, mut message: Message,
direction: Option<bool>, direction: Option<bool>,
) -> IFResult<()> { ) -> IFResult<()> {
let tunnel: Tunnel = if let Some(tun) = self.tunnels.iter().cloned().find(|t| t.id == Some(tunnel_id)) { tun } else { return Err(IFError::TunnelNotFound) }; let tunnel: Tunnel = if let Some(tun) = self
.tunnels
.iter()
.cloned()
.find(|t| t.id == Some(tunnel_id))
{
tun
} else {
return Err(IFError::TunnelNotFound);
};
message.tunnel_id = tunnel_id; message.tunnel_id = tunnel_id;
let peer_ids = match (direction, tunnel.peer_ids) { let peer_ids = match (direction, tunnel.peer_ids) {
(_, (x, 0)) => vec![x], (_, (x, 0)) => vec![x],
@ -120,10 +149,21 @@ impl IronForce {
if self.processed_messages.contains(&message.message_id) { if self.processed_messages.contains(&message.message_id) {
return Ok(()); return Ok(());
} }
#[cfg(feature = "std")]
println!("[{}] {:?}", self.short_id(), message);
self.processed_messages.push(message.message_id); self.processed_messages.push(message.message_id);
match &message.message_type { match &message.message_type {
MessageType::Service(msg_type) => match msg_type { MessageType::Service(msg_type) => {
match msg_type {
ServiceMessageType::TunnelBuildingForwardMovement(tunnel, sender_enc) => { ServiceMessageType::TunnelBuildingForwardMovement(tunnel, sender_enc) => {
let count = *self
.tunnel_counters
.get(&tunnel.local_ids[0])
.unwrap_or(&0u32);
if count > TUNNEL_MAX_REPEAT_COUNT {
return Ok(());
}
self.tunnel_counters.insert(tunnel.local_ids[0], count + 1);
if message.check_recipient(&self.keys) { if message.check_recipient(&self.keys) {
let mut tunnel_pub = tunnel.clone(); let mut tunnel_pub = tunnel.clone();
tunnel_pub.id = Some(rand::random()); tunnel_pub.id = Some(rand::random());
@ -135,31 +175,47 @@ impl IronForce {
ttd: 0, ttd: 0,
nodes_in_tunnel: None, nodes_in_tunnel: None,
is_multicast: false, is_multicast: false,
target_node: Some(sender.clone()), target_node: Some(sender),
}; };
#[cfg(feature = "std")]
println!(
"[{}] Got an incoming tunnel for me! {:?}",
self.short_id(),
tunnel_pub
);
self.tunnels.push(tunnel); self.tunnels.push(tunnel);
self.send_to_all( self.transport.send_message(
Message::build() serde_cbor::to_vec(
&Message::build()
.message_type(MessageType::Service( .message_type(MessageType::Service(
ServiceMessageType::TunnelBuildingBackwardMovement(tunnel_pub.clone()), ServiceMessageType::TunnelBuildingBackwardMovement(
tunnel_pub.clone(),
),
)) ))
.tunnel(tunnel_pub.id.unwrap()) .tunnel(tunnel_pub.id.unwrap())
.sign(&self.keys) .sign(&self.keys)
.build()? .build()?,
)?,
Some(inc_peer),
)?; )?;
} else { } else {
let mut tunnel = tunnel.clone(); let mut tunnel = tunnel.clone();
tunnel.local_ids.push(rand::random()); tunnel.add_local_id();
self.tunnels_pending.push((tunnel.clone(), None, (inc_peer, 0))); self.tunnels_pending
.push((tunnel.clone(), None, (inc_peer, 0)));
self.send_to_all(message)?; self.send_to_all(message)?;
} }
} }
ServiceMessageType::TunnelBuildingBackwardMovement(tunnel_p) => { ServiceMessageType::TunnelBuildingBackwardMovement(tunnel_p) => {
match self #[cfg(feature = "std")]
.tunnels_pending println!(
.iter() "[{}] Pending tunnels: {:?}",
.find(|tun| tunnel_p.local_ids.contains(tun.0.local_ids.last().unwrap())) self.short_id(),
{ self.tunnels_pending
);
match self.tunnels_pending.iter().find(|tun| {
tunnel_p.local_ids.contains(tun.0.local_ids.last().unwrap())
}) {
// This doesn't concern us // This doesn't concern us
None => {} None => {}
// This is a tunnel initialization proposed by us (and we got it back, yay) // This is a tunnel initialization proposed by us (and we got it back, yay)
@ -174,6 +230,8 @@ impl IronForce {
target_node: Some(target.clone()), target_node: Some(target.clone()),
}; };
self.tunnels.push(tunnel); self.tunnels.push(tunnel);
#[cfg(feature = "std")]
println!("[{}] Successfully created a new tunnel", self.short_id());
// Send some initialization message or something // Send some initialization message or something
} }
// This is a tunnel initialization proposed by someone else that has passed through us on its forward movement // This is a tunnel initialization proposed by someone else that has passed through us on its forward movement
@ -185,14 +243,18 @@ impl IronForce {
ttd: 0, ttd: 0,
nodes_in_tunnel: None, nodes_in_tunnel: None,
is_multicast: false, is_multicast: false,
target_node: None target_node: None,
}; };
self.tunnels.push(tunnel); self.tunnels.push(tunnel);
self.transport.send_message(serde_cbor::to_vec(&message)?, Some(peers.0))?; #[cfg(feature = "std")]
println!("[{}] Successfully created a new tunnel", self.short_id());
self.transport
.send_message(serde_cbor::to_vec(&message)?, Some(peers.0))?;
}
}
} }
} }
} }
},
MessageType::SingleCast | MessageType::Broadcast => self.messages.push(message), MessageType::SingleCast | MessageType::Broadcast => self.messages.push(message),
} }
Ok(()) Ok(())
@ -210,6 +272,19 @@ impl IronForce {
} }
Ok(()) Ok(())
} }
fn short_id(&self) -> alloc::string::String {
let vec_data = self.keys.get_public().to_vec();
alloc::string::String::from_utf8(
vec_data
.iter()
.skip(90)
.take(5)
.map(|c| c % 26 + 97)
.collect::<Vec<u8>>(),
)
.unwrap()
}
} }
#[cfg(test)] #[cfg(test)]
@ -217,12 +292,12 @@ mod if_testing {
use crate::crypto::Keys; use crate::crypto::Keys;
use crate::interface::test_interface::create_test_interfaces; use crate::interface::test_interface::create_test_interfaces;
use crate::ironforce::IronForce; use crate::ironforce::IronForce;
use crate::message::{Message, MessageType};
use crate::res::IFResult; use crate::res::IFResult;
use crate::transport::Transport; use crate::transport::Transport;
use alloc::boxed::Box; use alloc::boxed::Box;
use alloc::vec; use alloc::vec;
use alloc::vec::Vec; use alloc::vec::Vec;
use crate::message::{Message, MessageType};
fn create_test_network() -> Vec<IronForce> { fn create_test_network() -> Vec<IronForce> {
let interfaces = create_test_interfaces(5); let interfaces = create_test_interfaces(5);
@ -239,6 +314,7 @@ mod if_testing {
tunnels_pending: vec![], tunnels_pending: vec![],
has_background_worker: false, has_background_worker: false,
processed_messages: vec![], processed_messages: vec![],
tunnel_counters: Default::default(),
}) })
.collect() .collect()
} }
@ -271,11 +347,119 @@ mod if_testing {
.recipient(key_1.clone()) .recipient(key_1.clone())
.content(b"hello".to_vec()) .content(b"hello".to_vec())
.build()?, .build()?,
key_1)?; key_1,
)?;
network[1].main_loop_iteration()?; network[1].main_loop_iteration()?;
let msg = network[1].read_message(); let msg = network[1].read_message();
assert!(msg.is_some()); assert!(msg.is_some());
assert_eq!(msg.unwrap().get_decrypted(&network[1].keys).unwrap().as_slice(), b"hello"); assert_eq!(
msg.unwrap()
.get_decrypted(&network[1].keys)
.unwrap()
.as_slice(),
b"hello"
);
Ok(())
}
}
#[cfg(test)]
#[cfg(feature = "std")]
mod test_with_ip {
use crate::crypto::Keys;
use crate::interfaces::ip::IPInterface;
use crate::ironforce::IronForce;
use crate::res::IFResult;
use crate::transport::Transport;
use alloc::boxed::Box;
use alloc::vec;
use alloc::vec::Vec;
use core::str::FromStr;
use std::println;
fn create_test_interfaces(n: usize) -> impl Iterator<Item = IPInterface> {
let ip_addr = std::net::IpAddr::from_str("127.0.0.1").unwrap();
(0..n).map(move |i| {
IPInterface::new(
(5000 + 5 * i) as u16,
(0..n)
.filter(|j| *j != i)
.map(|j| (ip_addr, (5000 + 5 * j) as u16))
.collect(),
)
.unwrap()
})
}
fn create_test_network() -> Vec<IronForce> {
let interfaces = create_test_interfaces(4);
let transports = interfaces
.into_iter()
.map(|interface| Transport::new(vec![Box::new(interface)]));
transports
.map(|tr| IronForce {
keys: Keys::generate(),
transport: tr,
additional_modules: vec![],
tunnels: vec![],
messages: vec![],
tunnels_pending: vec![],
has_background_worker: false,
processed_messages: vec![],
tunnel_counters: Default::default()
})
.collect()
}
#[test]
fn test_creating_a_tunnel() -> IFResult<()> {
let mut network = create_test_network();
let key_1 = network[1].keys.get_public();
let (mut node0, mut node1) = (network.remove(0), network.remove(0));
println!("node0 id: {}", node0.short_id());
println!("node1 id: {}", node1.short_id());
let (mut node2, mut node3) = (network.remove(0), network.remove(0));
let t1 = std::thread::spawn(move || {
for _i in 0..5 {
// println!("Iteration {} (1)", i);
node0.main_loop_iteration().unwrap();
}
node0
});
let t2 = std::thread::spawn(move || {
for _i in 0..15 {
// println!("Iteration {} (2)", i);
node1.main_loop_iteration().unwrap();
}
node1
});
std::thread::spawn(move || loop {
node2.main_loop_iteration().unwrap();
std::thread::sleep(std::time::Duration::from_secs(5));
});
std::thread::spawn(move || loop {
std::thread::sleep(std::time::Duration::from_secs(5));
node3.main_loop_iteration().unwrap();
});
let mut node0 = t1.join().unwrap();
node0.initialize_tunnel_creation(key_1)?;
let mut node1 = t2.join().unwrap();
let t1 = std::thread::spawn(move || {
for _ in 0..18 {
node0.main_loop_iteration().unwrap();
std::thread::sleep(std::time::Duration::from_millis(150));
}
node0
});
let t2 = std::thread::spawn(move || {
for _ in 0..18 {
node1.main_loop_iteration().unwrap();
}
node1
});
let node0 = t1.join().unwrap();
t2.join().unwrap();
assert!(!node0.tunnels.is_empty());
Ok(()) Ok(())
} }
} }

2
src/res.rs

@ -9,6 +9,8 @@ pub enum IFError {
General(String), General(String),
/// A tunnel satisfying some conditions has not been found /// A tunnel satisfying some conditions has not been found
TunnelNotFound, TunnelNotFound,
/// Could not establish a connection
CouldNotConnect,
/// Error during serialization /// Error during serialization
SerializationError(String), SerializationError(String),
/// Error in rsa /// Error in rsa

19
src/tunnel.rs

@ -41,13 +41,15 @@ pub struct TunnelPublic {
impl TunnelPublic { impl TunnelPublic {
pub fn new_singlecast() -> Self { pub fn new_singlecast() -> Self {
Self { let mut tun = Self {
id: None, id: None,
local_ids: vec![rand::random()], local_ids: vec![],
ttd: 0, ttd: 0,
nodes_in_tunnel: None, nodes_in_tunnel: None,
is_multicast: false, is_multicast: false,
} };
tun.add_local_id();
tun
} }
/// Get the hash of the tunnel for verification /// Get the hash of the tunnel for verification
@ -67,6 +69,17 @@ impl TunnelPublic {
is_multicast: true, is_multicast: true,
} }
} }
pub fn add_local_id(&mut self) -> u64 {
let local_id = rand::random();
// Add 0 to 7 random ids so it's impossible to get the length of the tunnel
// todo: enable it (after debug)
// for _ in 0..(rand::random() as u64 % 7) {
// self.local_ids.push(rand::random())
// }
self.local_ids.push(local_id);
local_id
}
} }
#[test] #[test]

Loading…
Cancel
Save