Git-Mirrors/veilid
1
0
Fork 0
mirror of https://gitlab.com/veilid/veilid.git synced 2024-09-20 08:06:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch veilid:main into main

Browse Source
This commit is contained in:
Moritz Christian Weber 2023-08-26 21:17:41 +00:00
commit 187c86d9f7
19 changed files with 565 additions and 389 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
Cargo.lock generated
View File

@ -854,9 +854,9 @@ dependencies = [
[[package]]
name = "cc"
version = "1.0.83"
version = "1.0.82"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f1174fb0b6ec23863f8b971027804a42614e347eafb0a95bf0b12cdae21fc4d0"
checksum = "305fe645edc1442a0fa8b6726ba61d422798d37a52e12eaecf4b022ebbb88f01"
dependencies = [
"libc",
]
@ -1510,9 +1510,9 @@ dependencies = [
[[package]]
name = "dashmap"
version = "5.5.1"
version = "5.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "edd72493923899c6f10c641bdbdeddc7183d6396641d99c1a0d1597f37f92e28"
checksum = "6943ae99c34386c84a470c499d3414f66502a41340aa895406e0d2e4a207b91d"
dependencies = [
"cfg-if 1.0.0",
"hashbrown 0.14.0",
@ -2226,9 +2226,9 @@ dependencies = [
[[package]]
name = "h2"
version = "0.3.21"
version = "0.3.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "91fc23aa11be92976ef4729127f1a74adf36d8436f7816b185d18df956790833"
checksum = "97ec8491ebaf99c8eaa73058b045fe58073cd6be7f596ac993ced0b0a0c01049"
dependencies = [
"bytes 1.4.0",
"fnv",
@ -3728,12 +3728,12 @@ dependencies = [
[[package]]
name = "petgraph"
version = "0.6.4"
version = "0.6.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e1d3afd2628e69da2be385eb6f2fd57c8ac7977ceeff6dc166ff1657b0e386a9"
checksum = "4dd7d28ee937e54fe3080c91faa1c3a46c06de6252988a7f4592ba2310ef22a4"
dependencies = [
"fixedbitset",
"indexmap 2.0.0",
"indexmap 1.9.3",
]
[[package]]
@ -4434,9 +4434,9 @@ dependencies = [
[[package]]
name = "serde"
version = "1.0.185"
version = "1.0.183"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "be9b6f69f1dfd54c3b568ffa45c310d6973a5e5148fd40cf515acaf38cf5bc31"
checksum = "32ac8da02677876d532745a130fc9d8e6edfa81a269b107c5b00829b91d8eb3c"
dependencies = [
"serde_derive",
]
@ -4462,9 +4462,9 @@ dependencies = [
[[package]]
name = "serde_derive"
version = "1.0.185"
version = "1.0.183"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc59dfdcbad1437773485e0367fea4b090a2e0a16d9ffc46af47764536a298ec"
checksum = "aafe972d60b0b9bee71a91b92fee2d4fb3c9d7e8f6b179aa99f27203d99a4816"
dependencies = [
"proc-macro2",
"quote",
@ -4721,9 +4721,9 @@ dependencies = [
[[package]]
name = "slab"
version = "0.4.9"
version = "0.4.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f92a496fb766b417c996b9c5e57daf2f7ad3b0bebe1ccfca4856390e3d3bb67"
checksum = "6528351c9bc8ab22353f9d776db39a20288e8d6c37ef8cfe3317cf875eecfc2d"
dependencies = [
"autocfg",
]
@ -5699,6 +5699,7 @@ dependencies = [
"keyvaluedb-web",
"lazy_static",
"libc",
"lock_api",
"lz4_flex",
"ndk",
"ndk-glue",

2
veilid-core/src/lib.rs
View File

@ -90,7 +90,7 @@ pub static DEFAULT_LOG_IGNORE_LIST: [&str; 23] = [
use cfg_if::*;
use enumset::*;
use eyre::{bail, eyre, Report as EyreReport, Result as EyreResult, WrapErr};
use futures_util::stream::FuturesUnordered;
use futures_util::stream::{FuturesOrdered, FuturesUnordered};
use parking_lot::*;
use schemars::{schema_for, JsonSchema};
use serde::*;

199
veilid-core/src/network_manager/mod.rs
View File

@ -54,8 +54,8 @@ pub const IPADDR_TABLE_SIZE: usize = 1024;
pub const IPADDR_MAX_INACTIVE_DURATION_US: TimestampDuration =
TimestampDuration::new(300_000_000u64); // 5 minutes
pub const NODE_CONTACT_METHOD_CACHE_SIZE: usize = 1024;
pub const PUBLIC_ADDRESS_CHANGE_DETECTION_COUNT: usize = 3;
pub const PUBLIC_ADDRESS_CHECK_CACHE_SIZE: usize = 8;
pub const PUBLIC_ADDRESS_CHANGE_DETECTION_COUNT: usize = 5;
pub const PUBLIC_ADDRESS_CHECK_CACHE_SIZE: usize = 10;
pub const PUBLIC_ADDRESS_CHECK_TASK_INTERVAL_SECS: u32 = 60;
pub const PUBLIC_ADDRESS_INCONSISTENCY_TIMEOUT_US: TimestampDuration =
TimestampDuration::new(300_000_000u64); // 5 minutes
@ -1116,199 +1116,4 @@ impl NetworkManager {
// Inform caller that we dealt with the envelope locally
Ok(true)
}
// Determine if a local IP address has changed
// this means we should restart the low level network and and recreate all of our dial info
// Wait until we have received confirmation from N different peers
pub fn report_local_network_socket_address(
&self,
_socket_address: SocketAddress,
_connection_descriptor: ConnectionDescriptor,
_reporting_peer: NodeRef,
) {
// XXX: Nothing here yet.
}
// Determine if a global IP address has changed
// this means we should recreate our public dial info if it is not static and rediscover it
// Wait until we have received confirmation from N different peers
pub fn report_public_internet_socket_address(
&self,
socket_address: SocketAddress, // the socket address as seen by the remote peer
connection_descriptor: ConnectionDescriptor, // the connection descriptor used
reporting_peer: NodeRef, // the peer's noderef reporting the socket address
) {
#[cfg(feature = "verbose-tracing")]
debug!("report_global_socket_address\nsocket_address: {:#?}\nconnection_descriptor: {:#?}\nreporting_peer: {:#?}", socket_address, connection_descriptor, reporting_peer);
// Ignore these reports if we are currently detecting public dial info
let net = self.net();
if net.needs_public_dial_info_check() {
return;
}
let routing_table = self.routing_table();
let (detect_address_changes, ip6_prefix_size) = self.with_config(|c| {
(
c.network.detect_address_changes,
c.network.max_connections_per_ip6_prefix_size as usize,
)
});
// Get the ip(block) this report is coming from
let ipblock = ip_to_ipblock(
ip6_prefix_size,
connection_descriptor.remote_address().to_ip_addr(),
);
// Store the reported address if it isn't denylisted
let key = PublicAddressCheckCacheKey(
connection_descriptor.protocol_type(),
connection_descriptor.address_type(),
);
let mut inner = self.inner.lock();
let inner = &mut *inner;
let pacc = inner
.public_address_check_cache
.entry(key)
.or_insert_with(|| LruCache::new(PUBLIC_ADDRESS_CHECK_CACHE_SIZE));
let pait = inner
.public_address_inconsistencies_table
.entry(key)
.or_insert_with(|| HashMap::new());
if pait.contains_key(&ipblock) {
return;
}
pacc.insert(ipblock, socket_address, |_k, _v| {
// do nothing on LRU evict
});
// Determine if our external address has likely changed
let mut bad_public_address_detection_punishment: Option<
Box<dyn FnOnce() + Send + 'static>,
> = None;
let public_internet_network_class = routing_table
.get_network_class(RoutingDomain::PublicInternet)
.unwrap_or(NetworkClass::Invalid);
let needs_public_address_detection =
if matches!(public_internet_network_class, NetworkClass::InboundCapable) {
// Get the dial info filter for this connection so we can check if we have any public dialinfo that may have changed
let dial_info_filter = connection_descriptor.make_dial_info_filter();
// Get current external ip/port from registered global dialinfo
let current_addresses: BTreeSet<SocketAddress> = routing_table
.all_filtered_dial_info_details(
RoutingDomain::PublicInternet.into(),
&dial_info_filter,
)
.iter()
.map(|did| did.dial_info.socket_address())
.collect();
// If we are inbound capable, but start to see inconsistent socket addresses from multiple reporting peers
// then we zap the network class and re-detect it
let mut inconsistencies = Vec::new();
// Iteration goes from most recent to least recent node/address pair
for (reporting_ip_block, a) in pacc {
// If this address is not one of our current addresses (inconsistent)
// and we haven't already denylisted the reporting source,
if !current_addresses.contains(a) && !pait.contains_key(reporting_ip_block) {
// Record the origin of the inconsistency
inconsistencies.push(*reporting_ip_block);
}
}
// If we have enough inconsistencies to consider changing our public dial info,
// add them to our denylist (throttling) and go ahead and check for new
// public dialinfo
let inconsistent = if inconsistencies.len() >= PUBLIC_ADDRESS_CHANGE_DETECTION_COUNT
{
let exp_ts = get_aligned_timestamp() + PUBLIC_ADDRESS_INCONSISTENCY_TIMEOUT_US;
for i in &inconsistencies {
pait.insert(*i, exp_ts);
}
// Run this routine if the inconsistent nodes turn out to be lying
let this = self.clone();
bad_public_address_detection_punishment = Some(Box::new(move || {
let mut inner = this.inner.lock();
let pait = inner
.public_address_inconsistencies_table
.entry(key)
.or_insert_with(|| HashMap::new());
let exp_ts = get_aligned_timestamp()
+ PUBLIC_ADDRESS_INCONSISTENCY_PUNISHMENT_TIMEOUT_US;
for i in inconsistencies {
pait.insert(i, exp_ts);
}
}));
true
} else {
false
};
// // debug code
// if inconsistent {
// trace!("public_address_check_cache: {:#?}\ncurrent_addresses: {:#?}\ninconsistencies: {}", inner
// .public_address_check_cache, current_addresses, inconsistencies);
// }
inconsistent
} else if matches!(public_internet_network_class, NetworkClass::OutboundOnly) {
// If we are currently outbound only, we don't have any public dial info
// but if we are starting to see consistent socket address from multiple reporting peers
// then we may be become inbound capable, so zap the network class so we can re-detect it and any public dial info
let mut consistencies = 0;
let mut consistent = false;
let mut current_address = Option::<SocketAddress>::None;
// Iteration goes from most recent to least recent node/address pair
let pacc = inner
.public_address_check_cache
.entry(key)
.or_insert_with(|| LruCache::new(PUBLIC_ADDRESS_CHECK_CACHE_SIZE));
for (_, a) in pacc {
if let Some(current_address) = current_address {
if current_address == *a {
consistencies += 1;
if consistencies >= PUBLIC_ADDRESS_CHANGE_DETECTION_COUNT {
consistent = true;
break;
}
}
} else {
current_address = Some(*a);
}
}
consistent
} else {
// If we are a webapp we never do this.
// If we have invalid network class, then public address detection is already going to happen via the network_class_discovery task
false
};
if needs_public_address_detection {
if detect_address_changes {
// Reset the address check cache now so we can start detecting fresh
info!("Public address has changed, detecting public dial info");
inner.public_address_check_cache.clear();
// Re-detect the public dialinfo
net.set_needs_public_dial_info_check(bad_public_address_detection_punishment);
} else {
warn!("Public address may have changed. Restarting the server may be required.");
warn!("report_global_socket_address\nsocket_address: {:#?}\nconnection_descriptor: {:#?}\nreporting_peer: {:#?}", socket_address, connection_descriptor, reporting_peer);
warn!(
"public_address_check_cache: {:#?}",
inner.public_address_check_cache
);
}
}
}
}

340
veilid-core/src/network_manager/native/network_class_discovery.rs
View File

@ -1,3 +1,5 @@
/// Detect NetworkClass and DialInfo for the PublicInternet RoutingDomain
/// Also performs UPNP/IGD mapping if enabled and possible
use super::*;
use futures_util::stream::FuturesUnordered;
use futures_util::FutureExt;
@ -25,12 +27,20 @@ struct DiscoveryContextInner {
detected_public_dial_info: Option<DetectedPublicDialInfo>,
}
#[derive(Clone)]
pub struct DiscoveryContext {
routing_table: RoutingTable,
net: Network,
inner: Arc<Mutex<DiscoveryContextInner>>,
}
#[derive(Clone, Debug)]
struct DetectedDialInfo {
dial_info: DialInfo,
dial_info_class: DialInfoClass,
network_class: NetworkClass,
}
impl DiscoveryContext {
pub fn new(routing_table: RoutingTable, net: Network) -> Self {
Self {
@ -410,53 +420,74 @@ impl DiscoveryContext {
// If we know we are not behind NAT, check our firewall status
#[instrument(level = "trace", skip(self), err)]
pub async fn protocol_process_no_nat(&self) -> EyreResult<()> {
let (node_1, external_1_dial_info) = {
let inner = self.inner.lock();
(
inner.node_1.as_ref().unwrap().clone(),
inner.external_1_dial_info.as_ref().unwrap().clone(),
)
};
// Do these detections in parallel, but with ordering preference
let mut ord = FuturesOrdered::new();
// Attempt a port mapping via all available and enabled mechanisms
// Try this before the direct mapping in the event that we are restarting
// and may not have recorded a mapping created the last time
if let Some(external_mapped_dial_info) = self.try_port_mapping().await {
// Got a port mapping, let's use it
self.set_detected_public_dial_info(external_mapped_dial_info, DialInfoClass::Mapped);
self.set_detected_network_class(NetworkClass::InboundCapable);
}
// Do a validate_dial_info on the external address from a redirected node
else if self
.validate_dial_info(node_1.clone(), external_1_dial_info.clone(), true)
.await
{
// Add public dial info with Direct dialinfo class
self.set_detected_public_dial_info(external_1_dial_info, DialInfoClass::Direct);
self.set_detected_network_class(NetworkClass::InboundCapable);
} else {
// Add public dial info with Blocked dialinfo class
self.set_detected_public_dial_info(external_1_dial_info, DialInfoClass::Blocked);
self.set_detected_network_class(NetworkClass::InboundCapable);
// UPNP Automatic Mapping
///////////
let this = self.clone();
let do_mapped_fut: SendPinBoxFuture<Option<DetectedDialInfo>> = Box::pin(async move {
// Attempt a port mapping via all available and enabled mechanisms
// Try this before the direct mapping in the event that we are restarting
// and may not have recorded a mapping created the last time
if let Some(external_mapped_dial_info) = this.try_port_mapping().await {
// Got a port mapping, let's use it
return Some(DetectedDialInfo {
dial_info: external_mapped_dial_info.clone(),
dial_info_class: DialInfoClass::Mapped,
network_class: NetworkClass::InboundCapable,
});
}
None
});
ord.push_back(do_mapped_fut);
let this = self.clone();
let do_direct_fut: SendPinBoxFuture<Option<DetectedDialInfo>> = Box::pin(async move {
let (node_1, external_1_dial_info) = {
let inner = this.inner.lock();
(
inner.node_1.as_ref().unwrap().clone(),
inner.external_1_dial_info.as_ref().unwrap().clone(),
)
};
// Do a validate_dial_info on the external address from a redirected node
if this
.validate_dial_info(node_1.clone(), external_1_dial_info.clone(), true)
.await
{
// Add public dial info with Direct dialinfo class
Some(DetectedDialInfo {
dial_info: external_1_dial_info.clone(),
dial_info_class: DialInfoClass::Direct,
network_class: NetworkClass::InboundCapable,
})
} else {
// Add public dial info with Blocked dialinfo class
Some(DetectedDialInfo {
dial_info: external_1_dial_info.clone(),
dial_info_class: DialInfoClass::Blocked,
network_class: NetworkClass::InboundCapable,
})
}
});
ord.push_back(do_direct_fut);
while let Some(res) = ord.next().await {
if let Some(ddi) = res {
self.set_detected_public_dial_info(ddi.dial_info, ddi.dial_info_class);
self.set_detected_network_class(ddi.network_class);
break;
}
}
Ok(())
}
// If we know we are behind NAT check what kind
#[instrument(level = "trace", skip(self), ret, err)]
pub async fn protocol_process_nat(&self) -> EyreResult<bool> {
// Attempt a port mapping via all available and enabled mechanisms
// Try this before the direct mapping in the event that we are restarting
// and may not have recorded a mapping created the last time
if let Some(external_mapped_dial_info) = self.try_port_mapping().await {
// Got a port mapping, let's use it
self.set_detected_public_dial_info(external_mapped_dial_info, DialInfoClass::Mapped);
self.set_detected_network_class(NetworkClass::InboundCapable);
// No more retries
return Ok(true);
}
// Get the external dial info for our use here
let (node_1, external_1_dial_info, external_1_address, protocol_type, address_type) = {
let inner = self.inner.lock();
@ -469,36 +500,111 @@ impl DiscoveryContext {
)
};
// Do a validate_dial_info on the external address from a redirected node
if self
.validate_dial_info(node_1.clone(), external_1_dial_info.clone(), true)
.await
{
// Add public dial info with Direct dialinfo class
self.set_detected_public_dial_info(external_1_dial_info, DialInfoClass::Direct);
self.set_detected_network_class(NetworkClass::InboundCapable);
return Ok(true);
// Do these detections in parallel, but with ordering preference
let mut ord = FuturesOrdered::new();
// UPNP Automatic Mapping
///////////
let this = self.clone();
let do_mapped_fut: SendPinBoxFuture<Option<DetectedDialInfo>> = Box::pin(async move {
// Attempt a port mapping via all available and enabled mechanisms
// Try this before the direct mapping in the event that we are restarting
// and may not have recorded a mapping created the last time
if let Some(external_mapped_dial_info) = this.try_port_mapping().await {
// Got a port mapping, let's use it
return Some(DetectedDialInfo {
dial_info: external_mapped_dial_info.clone(),
dial_info_class: DialInfoClass::Mapped,
network_class: NetworkClass::InboundCapable,
});
}
None
});
ord.push_back(do_mapped_fut);
// Manual Mapping Detection
///////////
let this = self.clone();
if let Some(local_port) = this.net.get_local_port(protocol_type) {
if external_1_dial_info.port() != local_port {
let c_external_1_dial_info = external_1_dial_info.clone();
let c_node_1 = node_1.clone();
let do_manual_map_fut: SendPinBoxFuture<Option<DetectedDialInfo>> =
Box::pin(async move {
// Do a validate_dial_info on the external address, but with the same port as the local port of local interface, from a redirected node
// This test is to see if a node had manual port forwarding done with the same port number as the local listener
let mut external_1_dial_info_with_local_port =
c_external_1_dial_info.clone();
external_1_dial_info_with_local_port.set_port(local_port);
if this
.validate_dial_info(
c_node_1.clone(),
external_1_dial_info_with_local_port.clone(),
true,
)
.await
{
// Add public dial info with Direct dialinfo class
return Some(DetectedDialInfo {
dial_info: external_1_dial_info_with_local_port,
dial_info_class: DialInfoClass::Direct,
network_class: NetworkClass::InboundCapable,
});
}
None
});
ord.push_back(do_manual_map_fut);
}
}
// Port mapping was not possible, and things aren't accessible directly.
// Let's see what kind of NAT we have
// Full Cone NAT Detection
///////////
let this = self.clone();
let c_node_1 = node_1.clone();
let c_external_1_dial_info = external_1_dial_info.clone();
let do_full_cone_fut: SendPinBoxFuture<Option<DetectedDialInfo>> = Box::pin(async move {
// Let's see what kind of NAT we have
// Does a redirected dial info validation from a different address and a random port find us?
if this
.validate_dial_info(c_node_1.clone(), c_external_1_dial_info.clone(), true)
.await
{
// Yes, another machine can use the dial info directly, so Full Cone
// Add public dial info with full cone NAT network class
// Does a redirected dial info validation from a different address and a random port find us?
if self
.validate_dial_info(node_1.clone(), external_1_dial_info.clone(), true)
.await
{
// Yes, another machine can use the dial info directly, so Full Cone
// Add public dial info with full cone NAT network class
self.set_detected_public_dial_info(external_1_dial_info, DialInfoClass::FullConeNAT);
self.set_detected_network_class(NetworkClass::InboundCapable);
return Some(DetectedDialInfo {
dial_info: c_external_1_dial_info,
dial_info_class: DialInfoClass::FullConeNAT,
network_class: NetworkClass::InboundCapable,
});
}
None
});
ord.push_back(do_full_cone_fut);
// No more retries
return Ok(true);
// Run detections in parallel and take the first one, ordered by preference, that returns a result
while let Some(res) = ord.next().await {
if let Some(ddi) = res {
self.set_detected_public_dial_info(ddi.dial_info, ddi.dial_info_class);
self.set_detected_network_class(ddi.network_class);
return Ok(true);
}
}
// No, we are restricted, determine what kind of restriction
// We are restricted, determine what kind of restriction
// Get the external dial info for our use here
let (node_1, external_1_dial_info, external_1_address, protocol_type, address_type) = {
let inner = self.inner.lock();
(
inner.node_1.as_ref().unwrap().clone(),
inner.external_1_dial_info.as_ref().unwrap().clone(),
inner.external_1_address.unwrap(),
inner.protocol_type.unwrap(),
inner.address_type.unwrap(),
)
};
// Get our external address from some fast node, that is not node 1, call it node 2
let (external_2_address, node_2) = match self
.discover_external_address(protocol_type, address_type, Some(node_1.node_ids()))
@ -555,85 +661,25 @@ impl DiscoveryContext {
impl Network {
#[instrument(level = "trace", skip(self, context), err)]
pub async fn update_ipv4_protocol_dialinfo(
pub async fn update_protocol_dialinfo(
&self,
context: &DiscoveryContext,
protocol_type: ProtocolType,
address_type: AddressType,
) -> EyreResult<()> {
let mut retry_count = {
let c = self.config.get();
c.network.restricted_nat_retries
};
// Start doing ipv4 protocol
context.protocol_begin(protocol_type, AddressType::IPV4);
// Start doing protocol
context.protocol_begin(protocol_type, address_type);
// Loop for restricted NAT retries
loop {
log_net!(debug
"=== update_ipv4_protocol_dialinfo {:?} tries_left={} ===",
protocol_type,
retry_count
);
// Get our external address from some fast node, call it node 1
if !context.protocol_get_external_address_1().await {
// If we couldn't get an external address, then we should just try the whole network class detection again later
return Ok(());
}
// If our local interface list contains external_1 then there is no NAT in place
{
let res = {
let inner = context.inner.lock();
inner
.intf_addrs
.as_ref()
.unwrap()
.contains(inner.external_1_address.as_ref().unwrap())
};
if res {
// No NAT
context.protocol_process_no_nat().await?;
// No more retries
break;
}
}
// There is -some NAT-
if context.protocol_process_nat().await? {
// We either got dial info or a network class without one
break;
}
// If we tried everything, break anyway after N attempts
if retry_count == 0 {
break;
}
retry_count -= 1;
}
Ok(())
}
#[instrument(level = "trace", skip(self, context), err)]
pub async fn update_ipv6_protocol_dialinfo(
&self,
context: &DiscoveryContext,
protocol_type: ProtocolType,
) -> EyreResult<()> {
let mut retry_count = {
let c = self.config.get();
c.network.restricted_nat_retries
};
// Start doing ipv6 protocol
context.protocol_begin(protocol_type, AddressType::IPV6);
// Loop for restricted NAT retries
loop {
log_net!(debug
"=== update_ipv6_protocol_dialinfo {:?} tries_left={} ===",
"=== update_protocol_dialinfo {:?} {:?} tries_left={} ===",
address_type,
protocol_type,
retry_count
);
@ -714,7 +760,11 @@ impl Network {
let udpv4_context =
DiscoveryContext::new(self.routing_table(), self.clone());
if let Err(e) = self
.update_ipv4_protocol_dialinfo(&udpv4_context, ProtocolType::UDP)
.update_protocol_dialinfo(
&udpv4_context,
ProtocolType::UDP,
AddressType::IPV4,
)
.await
{
log_net!(debug "Failed UDPv4 dialinfo discovery: {}", e);
@ -734,7 +784,11 @@ impl Network {
let udpv6_context =
DiscoveryContext::new(self.routing_table(), self.clone());
if let Err(e) = self
.update_ipv6_protocol_dialinfo(&udpv6_context, ProtocolType::UDP)
.update_protocol_dialinfo(
&udpv6_context,
ProtocolType::UDP,
AddressType::IPV6,
)
.await
{
log_net!(debug "Failed UDPv6 dialinfo discovery: {}", e);
@ -757,7 +811,11 @@ impl Network {
let tcpv4_context =
DiscoveryContext::new(self.routing_table(), self.clone());
if let Err(e) = self
.update_ipv4_protocol_dialinfo(&tcpv4_context, ProtocolType::TCP)
.update_protocol_dialinfo(
&tcpv4_context,
ProtocolType::TCP,
AddressType::IPV4,
)
.await
{
log_net!(debug "Failed TCPv4 dialinfo discovery: {}", e);
@ -777,7 +835,11 @@ impl Network {
let wsv4_context =
DiscoveryContext::new(self.routing_table(), self.clone());
if let Err(e) = self
.update_ipv4_protocol_dialinfo(&wsv4_context, ProtocolType::WS)
.update_protocol_dialinfo(
&wsv4_context,
ProtocolType::WS,
AddressType::IPV4,
)
.await
{
log_net!(debug "Failed WSv4 dialinfo discovery: {}", e);
@ -800,7 +862,11 @@ impl Network {
let tcpv6_context =
DiscoveryContext::new(self.routing_table(), self.clone());
if let Err(e) = self
.update_ipv6_protocol_dialinfo(&tcpv6_context, ProtocolType::TCP)
.update_protocol_dialinfo(
&tcpv6_context,
ProtocolType::TCP,
AddressType::IPV6,
)
.await
{
log_net!(debug "Failed TCPv6 dialinfo discovery: {}", e);
@ -820,7 +886,11 @@ impl Network {
let wsv6_context =
DiscoveryContext::new(self.routing_table(), self.clone());
if let Err(e) = self
.update_ipv6_protocol_dialinfo(&wsv6_context, ProtocolType::WS)
.update_protocol_dialinfo(
&wsv6_context,
ProtocolType::WS,
AddressType::IPV6,
)
.await
{
log_net!(debug "Failed WSv6 dialinfo discovery: {}", e);

6
veilid-core/src/network_manager/native/protocol/udp.rs
View File

@ -79,13 +79,13 @@ impl RawUdpProtocolHandler {
};
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.len", &size);
tracing::Span::current().record("ret.len", &message_len);
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.descriptor", &format!("{:?}", descriptor).as_str());
Ok((message_len, descriptor))
}
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", err, skip(self, data), fields(data.len = data.len(), ret.len, ret.descriptor)))]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", err, skip(self, data), fields(data.len = data.len(), ret.descriptor)))]
pub async fn send_message(
&self,
data: Vec<u8>,
@ -133,8 +133,6 @@ impl RawUdpProtocolHandler {
SocketAddress::from_socket_addr(local_socket_addr),
);
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.len", &len);
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.descriptor", &format!("{:?}", descriptor).as_str());
Ok(NetworkResult::value(descriptor))

246
veilid-core/src/network_manager/tasks/public_address_check.rs
View File

@ -24,4 +24,250 @@ impl NetworkManager {
}
Ok(())
}
// Determine if a local IP address has changed
// this means we should restart the low level network and and recreate all of our dial info
// Wait until we have received confirmation from N different peers
pub fn report_local_network_socket_address(
&self,
_socket_address: SocketAddress,
_connection_descriptor: ConnectionDescriptor,
_reporting_peer: NodeRef,
) {
// XXX: Nothing here yet.
}
// Determine if a global IP address has changed
// this means we should recreate our public dial info if it is not static and rediscover it
// Wait until we have received confirmation from N different peers
pub fn report_public_internet_socket_address(
&self,
socket_address: SocketAddress, // the socket address as seen by the remote peer
connection_descriptor: ConnectionDescriptor, // the connection descriptor used
reporting_peer: NodeRef, // the peer's noderef reporting the socket address
) {
#[cfg(feature = "network-result-extra")]
debug!("report_global_socket_address\nsocket_address: {:#?}\nconnection_descriptor: {:#?}\nreporting_peer: {:#?}", socket_address, connection_descriptor, reporting_peer);
// Ignore these reports if we are currently detecting public dial info
let net = self.net();
if net.needs_public_dial_info_check() {
return;
}
// If we are a webapp we should skip this completely
// because we will never get inbound dialinfo directly on our public ip address
// If we have an invalid network class, this is not necessary yet
let routing_table = self.routing_table();
let public_internet_network_class = routing_table
.get_network_class(RoutingDomain::PublicInternet)
.unwrap_or(NetworkClass::Invalid);
if matches!(
public_internet_network_class,
NetworkClass::Invalid | NetworkClass::WebApp
) {
return;
}
let (detect_address_changes, ip6_prefix_size) = self.with_config(|c| {
(
c.network.detect_address_changes,
c.network.max_connections_per_ip6_prefix_size as usize,
)
});
// Get the ip(block) this report is coming from
let ipblock = ip_to_ipblock(
ip6_prefix_size,
connection_descriptor.remote_address().to_ip_addr(),
);
// Reject public address reports from nodes that we know are behind symmetric nat or
// nodes that must be using a relay for everything
let Some(node_info) = reporting_peer.node_info(RoutingDomain::PublicInternet) else {
return;
};
if node_info.network_class() != NetworkClass::InboundCapable {
return;
}
// Check if the public address report is coming from a node/block that gives an 'inconsistent' location
// meaning that the node may be not useful for public address detection
// This is done on a per address/protocol basis
let mut inner = self.inner.lock();
let inner = &mut *inner;
let addr_proto_type_key = PublicAddressCheckCacheKey(
connection_descriptor.protocol_type(),
connection_descriptor.address_type(),
);
if inner
.public_address_inconsistencies_table
.get(&addr_proto_type_key)
.map(|pait| pait.contains_key(&ipblock))
.unwrap_or(false)
{
return;
}
// Insert this new public address into the lru cache for the address check
// if we've seen this address before, it brings it to the front
let pacc = inner
.public_address_check_cache
.entry(addr_proto_type_key)
.or_insert_with(|| LruCache::new(PUBLIC_ADDRESS_CHECK_CACHE_SIZE));
pacc.insert(ipblock, socket_address, |_k, _v| {
// do nothing on LRU evict
});
// Determine if our external address has likely changed
let mut bad_public_address_detection_punishment: Option<
Box<dyn FnOnce() + Send + 'static>,
> = None;
let needs_public_address_detection = if matches!(
public_internet_network_class,
NetworkClass::InboundCapable
) {
// Get the dial info filter for this connection so we can check if we have any public dialinfo that may have changed
let dial_info_filter = connection_descriptor.make_dial_info_filter();
// Get current external ip/port from registered global dialinfo
let current_addresses: BTreeSet<SocketAddress> = routing_table
.all_filtered_dial_info_details(
RoutingDomain::PublicInternet.into(),
&dial_info_filter,
)
.iter()
.map(|did| {
// Strip port from direct and mapped addresses
// as the incoming dialinfo may not match the outbound
// connections' NAT mapping. In this case we only check for IP address changes.
if did.class == DialInfoClass::Direct || did.class == DialInfoClass::Mapped {
did.dial_info.socket_address().with_port(0)
} else {
did.dial_info.socket_address()
}
})
.collect();
// If we are inbound capable, but start to see inconsistent socket addresses from multiple reporting peers
// then we zap the network class and re-detect it
// Keep list of the origin ip blocks of inconsistent public address reports
let mut inconsistencies = Vec::new();
// Iteration goes from most recent to least recent node/address pair
for (reporting_ip_block, a) in pacc {
// If this address is not one of our current addresses (inconsistent)
// and we haven't already denylisted the reporting source,
// Also check address with port zero in the even we are only checking changes to ip addresses
if !current_addresses.contains(a)
&& !current_addresses.contains(&a.with_port(0))
&& !inner
.public_address_inconsistencies_table
.get(&addr_proto_type_key)
.map(|pait| pait.contains_key(reporting_ip_block))
.unwrap_or(false)
{
// Record the origin of the inconsistency
#[cfg(feature = "network-result-extra")]
debug!("inconsistency added from {:?}: reported {:?} with current_addresses = {:?}", reporting_ip_block, a, current_addresses);
inconsistencies.push(*reporting_ip_block);
}
}
// If we have enough inconsistencies to consider changing our public dial info,
// add them to our denylist (throttling) and go ahead and check for new
// public dialinfo
let inconsistent = if inconsistencies.len() >= PUBLIC_ADDRESS_CHANGE_DETECTION_COUNT {
let exp_ts = get_aligned_timestamp() + PUBLIC_ADDRESS_INCONSISTENCY_TIMEOUT_US;
let pait = inner
.public_address_inconsistencies_table
.entry(addr_proto_type_key)
.or_insert_with(|| HashMap::new());
for i in &inconsistencies {
pait.insert(*i, exp_ts);
}
// Run this routine if the inconsistent nodes turn out to be lying
let this = self.clone();
bad_public_address_detection_punishment = Some(Box::new(move || {
let mut inner = this.inner.lock();
let pait = inner
.public_address_inconsistencies_table
.entry(addr_proto_type_key)
.or_insert_with(|| HashMap::new());
let exp_ts = get_aligned_timestamp()
+ PUBLIC_ADDRESS_INCONSISTENCY_PUNISHMENT_TIMEOUT_US;
for i in inconsistencies {
pait.insert(i, exp_ts);
}
}));
true
} else {
false
};
// // debug code
// if inconsistent {
// trace!("public_address_check_cache: {:#?}\ncurrent_addresses: {:#?}\ninconsistencies: {}", inner
// .public_address_check_cache, current_addresses, inconsistencies);
// }
inconsistent
} else if matches!(public_internet_network_class, NetworkClass::OutboundOnly) {
// If we are currently outbound only, we don't have any public dial info
// but if we are starting to see consistent socket address from multiple reporting peers
// then we may be become inbound capable, so zap the network class so we can re-detect it and any public dial info
let mut consistencies = 0;
let mut consistent = false;
let mut current_address = Option::<SocketAddress>::None;
// Iteration goes from most recent to least recent node/address pair
for (_, a) in pacc {
if let Some(current_address) = current_address {
if current_address == *a {
consistencies += 1;
if consistencies >= PUBLIC_ADDRESS_CHANGE_DETECTION_COUNT {
consistent = true;
break;
}
}
} else {
current_address = Some(*a);
}
}
consistent
} else {
// If we are a webapp we never do this.
// If we have invalid network class, then public address detection is already going to happen via the network_class_discovery task
// we should have checked for this condition earlier at the top of this function
unreachable!();
};
if needs_public_address_detection {
if detect_address_changes {
// Reset the address check cache now so we can start detecting fresh
info!("Public address has changed, detecting public dial info");
inner.public_address_check_cache.clear();
// Re-detect the public dialinfo
net.set_needs_public_dial_info_check(bad_public_address_detection_punishment);
} else {
warn!("Public address may have changed. Restarting the server may be required.");
warn!("report_global_socket_address\nsocket_address: {:#?}\nconnection_descriptor: {:#?}\nreporting_peer: {:#?}", socket_address, connection_descriptor, reporting_peer);
warn!(
"public_address_check_cache: {:#?}",
inner.public_address_check_cache
);
}
}
}
}

4
veilid-core/src/network_manager/types/dial_info_class.rs
View File

@ -5,8 +5,8 @@ use super::*;
pub enum DialInfoClass {
Direct = 0, // D = Directly reachable with public IP and no firewall, with statically configured port
Mapped = 1, // M = Directly reachable with via portmap behind any NAT or firewalled with dynamically negotiated port
FullConeNAT = 2, // F = Directly reachable device without portmap behind full-cone NAT
Blocked = 3, // B = Inbound blocked at firewall but may hole punch with public address
FullConeNAT = 2, // F = Directly reachable device without portmap behind full-cone NAT (or manually mapped firewall port with no configuration change)
Blocked = 3, // B = Inbound blocked at firewall but may hole punch with public address
AddressRestrictedNAT = 4, // A = Device without portmap behind address-only restricted NAT
PortRestrictedNAT = 5, // P = Device without portmap behind address-and-port restricted NAT
}

5
veilid-core/src/network_manager/types/network_class.rs
View File

@ -15,6 +15,11 @@ impl Default for NetworkClass {
}
impl NetworkClass {
// Must an inbound relay be kept available?
// In the case of InboundCapable, it is left up to the class of each DialInfo to determine if an inbound relay is required
pub fn inbound_wants_relay(&self) -> bool {
matches!(self, Self::OutboundOnly | Self::WebApp)
}
// Should an outbound relay be kept available?
pub fn outbound_wants_relay(&self) -> bool {
matches!(self, Self::WebApp)

5
veilid-core/src/network_manager/types/socket_address.rs
View File

@ -33,6 +33,11 @@ impl SocketAddress {
pub fn set_port(&mut self, port: u16) {
self.port = port
}
pub fn with_port(&self, port: u16) -> Self {
let mut sa = self.clone();
sa.port = port;
sa
}
pub fn to_canonical(&self) -> SocketAddress {
SocketAddress {
address: self.address.to_canonical(),

4
veilid-core/src/routing_table/mod.rs
View File

@ -527,10 +527,6 @@ impl RoutingTable {
}
/// Look up the best way for two nodes to reach each other over a specific routing domain
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self), ret)
)]
pub fn get_contact_method(
&self,
routing_domain: RoutingDomain,

21
veilid-core/src/routing_table/routing_domains.rs
View File

@ -128,7 +128,21 @@ impl RoutingDomainDetailCommon {
self.dial_info_details.clone()
);
let relay_info = self
// Check if any of our dialinfo require a relay for signaling
// FullConeNAT requires a relay but it does not have to be published because it does not require signaling
let mut publish_relay = node_info.network_class().inbound_wants_relay() || node_info.network_class().outbound_wants_relay();
if !publish_relay {
// Check the dialinfo to see if they might want to publish a relay for signalling specifically
for did in self.dial_info_details() {
if did.class.requires_signal() {
publish_relay = true;
break;
}
}
}
let relay_info = if publish_relay {
self
.relay_node
.as_ref()
.and_then(|rn| {
@ -145,7 +159,10 @@ impl RoutingDomainDetailCommon {
} else {
None
}
});
})
} else {
None
};
let signed_node_info = match relay_info {
Some((relay_ids, relay_sdni)) => SignedNodeInfo::Relayed(

4
veilid-core/src/routing_table/routing_table_inner.rs
View File

@ -215,10 +215,6 @@ impl RoutingTableInner {
true
}
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self), ret)
)]
pub fn get_contact_method(
&self,
routing_domain: RoutingDomain,

19
veilid-flutter/example/ios/Podfile.lock
View File

@ -1,28 +1,35 @@
PODS:
- Flutter (1.0.0)
- path_provider_ios (0.0.1):
- path_provider_foundation (0.0.1):
- Flutter
- FlutterMacOS
- system_info_plus (0.0.1):
- Flutter
- veilid (0.0.1):
- Flutter
DEPENDENCIES:
- Flutter (from `Flutter`)
- path_provider_ios (from `.symlinks/plugins/path_provider_ios/ios`)
- path_provider_foundation (from `.symlinks/plugins/path_provider_foundation/darwin`)
- system_info_plus (from `.symlinks/plugins/system_info_plus/ios`)
- veilid (from `.symlinks/plugins/veilid/ios`)
EXTERNAL SOURCES:
Flutter:
:path: Flutter
path_provider_ios:
:path: ".symlinks/plugins/path_provider_ios/ios"
path_provider_foundation:
:path: ".symlinks/plugins/path_provider_foundation/darwin"
system_info_plus:
:path: ".symlinks/plugins/system_info_plus/ios"
veilid:
:path: ".symlinks/plugins/veilid/ios"
SPEC CHECKSUMS:
Flutter: f04841e97a9d0b0a8025694d0796dd46242b2854
path_provider_ios: 14f3d2fd28c4fdb42f44e0f751d12861c43cee02
path_provider_foundation: 29f094ae23ebbca9d3d0cec13889cd9060c0e943
system_info_plus: 5393c8da281d899950d751713575fbf91c7709aa
veilid: f5c2e662f91907b30cf95762619526ac3e4512fd
PODFILE CHECKSUM: ef19549a9bc3046e7bb7d2fab4d021637c0c58a3
COCOAPODS: 1.11.3
COCOAPODS: 1.12.1

5
veilid-flutter/example/ios/Runner.xcodeproj/project.pbxproj
View File

@ -157,7 +157,7 @@
97C146E61CF9000F007C117D /* Project object */ = {
isa = PBXProject;
attributes = {
LastUpgradeCheck = 1300;
LastUpgradeCheck = 1430;
ORGANIZATIONNAME = "";
TargetAttributes = {
97C146ED1CF9000F007C117D = {
@ -361,6 +361,7 @@
ASSETCATALOG_COMPILER_APPICON_NAME = AppIcon;
CLANG_ENABLE_MODULES = YES;
CURRENT_PROJECT_VERSION = "$(FLUTTER_BUILD_NUMBER)";
DEVELOPMENT_TEAM = XP5LBLT7M7;
ENABLE_BITCODE = NO;
INFOPLIST_FILE = Runner/Info.plist;
LD_RUNPATH_SEARCH_PATHS = (
@ -491,6 +492,7 @@
ASSETCATALOG_COMPILER_APPICON_NAME = AppIcon;
CLANG_ENABLE_MODULES = YES;
CURRENT_PROJECT_VERSION = "$(FLUTTER_BUILD_NUMBER)";
DEVELOPMENT_TEAM = XP5LBLT7M7;
ENABLE_BITCODE = NO;
INFOPLIST_FILE = Runner/Info.plist;
LD_RUNPATH_SEARCH_PATHS = (
@ -513,6 +515,7 @@
ASSETCATALOG_COMPILER_APPICON_NAME = AppIcon;
CLANG_ENABLE_MODULES = YES;
CURRENT_PROJECT_VERSION = "$(FLUTTER_BUILD_NUMBER)";
DEVELOPMENT_TEAM = XP5LBLT7M7;
ENABLE_BITCODE = NO;
INFOPLIST_FILE = Runner/Info.plist;
LD_RUNPATH_SEARCH_PATHS = (

2
veilid-flutter/example/ios/Runner.xcodeproj/xcshareddata/xcschemes/Runner.xcscheme
View File

@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<Scheme
LastUpgradeVersion = "1300"
LastUpgradeVersion = "1430"
version = "1.3">
<BuildAction
parallelizeBuildables = "YES"

2
veilid-flutter/example/macos/Runner.xcodeproj/project.pbxproj
View File

@ -208,7 +208,7 @@
isa = PBXProject;
attributes = {
LastSwiftUpdateCheck = 0920;
LastUpgradeCheck = 1300;
LastUpgradeCheck = 1430;
ORGANIZATIONNAME = "";
TargetAttributes = {
33CC10EC2044A3C60003C045 = {

2
veilid-flutter/example/macos/Runner.xcodeproj/xcshareddata/xcschemes/Runner.xcscheme
View File

@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<Scheme
LastUpgradeVersion = "1300"
LastUpgradeVersion = "1430"
version = "1.3">
<BuildAction
parallelizeBuildables = "YES"

30
veilid-flutter/rust/Cargo.toml
View File

@ -9,20 +9,36 @@ edition = "2021"
crate-type = ["cdylib", "staticlib", "rlib"]
[features]
default = [ "rt-tokio", "veilid-core/default" ]
crypto-test = [ "rt-tokio", "veilid-core/crypto-test"]
rt-async-std = [ "veilid-core/rt-async-std", "async-std", "opentelemetry/rt-async-std", "opentelemetry-otlp/grpc-sys"]
rt-tokio = [ "veilid-core/rt-tokio", "tokio", "tokio-stream", "tokio-util", "opentelemetry/rt-tokio"]
default = ["rt-tokio", "veilid-core/default"]
crypto-test = ["rt-tokio", "veilid-core/crypto-test"]
rt-async-std = [
"veilid-core/rt-async-std",
"async-std",
"opentelemetry/rt-async-std",
"opentelemetry-otlp/grpc-sys",
]
rt-tokio = [
"veilid-core/rt-tokio",
"tokio",
"tokio-stream",
"tokio-util",
"opentelemetry/rt-tokio",
]
[dependencies]
veilid-core = { path="../../veilid-core", default-features = false }
veilid-core = { path = "../../veilid-core", default-features = false, features = [
"verbose-tracing",
"network-result-extra",
] }
tracing = { version = "^0", features = ["log", "attributes"] }
tracing-subscriber = "^0"
parking_lot = "^0"
backtrace = "^0"
serde_json = "^1"
serde = "^1"
futures-util = { version = "^0", default_features = false, features = ["alloc"] }
futures-util = { version = "^0", default_features = false, features = [
"alloc",
] }
cfg-if = "^1"
data-encoding = { version = "^2" }
@ -36,7 +52,7 @@ opentelemetry-semantic-conventions = "0.10"
async-std = { version = "^1", features = ["unstable"], optional = true }
tokio = { version = "^1", features = ["full"], optional = true }
tokio-stream = { version = "^0", features = ["net"], optional = true }
tokio-util = { version = "^0", features = ["compat"], optional = true}
tokio-util = { version = "^0", features = ["compat"], optional = true }
allo-isolate = "^0"
ffi-support = "^0"
lazy_static = "^1"

27
veilid-python/update_schema.sh
View File

@ -2,16 +2,27 @@
set -eo pipefail
SCRIPTDIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
VEILID_SERVER=$SCRIPTDIR/../target/debug/veilid-server
# If $VEILID_SERVER is set, use that, otherwise find a valid Veilid server by looking in the usual places
if [ -z "${VEILID_SERVER}" ]; then
for VEILID_SERVER_CANDIDATE in ${SCRIPTDIR}/../target/debug/veilid-server; do
echo -n "Trying Veilid server at ${VEILID_SERVER_CANDIDATE}..."
if [ -f "${VEILID_SERVER_CANDIDATE}" ]; then
echo " found!"
VEILID_SERVER="${VEILID_SERVER_CANDIDATE}"
break
else
echo " not found."
fi
done
fi
# Ensure executable exists
if [ ! -f "$VEILID_SERVER" ]; then
echo "$VEILID_SERVER does not exist. Build with cargo build."
# If $VEILID_SERVER is still not set, or if it doesn't actually exist, bail
if [[ -z "${VEILID_SERVER}" || ! -f "${VEILID_SERVER}" ]]; then
echo "No valid in-tree Veilid server was found. Go to the top level directory, run 'cargo build', then change back to this directory and run this script again."
exit 1
fi
# Produce schema from veilid-server
$VEILID_SERVER --emit-schema Request > $SCRIPTDIR/veilid/schema/Request.json
$VEILID_SERVER --emit-schema RecvMessage > $SCRIPTDIR/veilid/schema/RecvMessage.json
for SCHEMA in "Request" "RecvMessage"; do
echo -n "Updating ${SCHEMA}..." && ${VEILID_SERVER} --emit-schema ${SCHEMA} > $SCRIPTDIR/veilid/schema/${SCHEMA} && echo " done." || echo " error!"
done