liz/veilid
liz/veilid
1
0
Fork 0
Code Pull Requests Packages Releases Activity

checkpoint

Browse Source
This commit is contained in:
John Smith 2023-02-20 20:37:52 -05:00
parent 05be3099c8
commit ae991334d3
25 changed files with 357 additions and 163 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
veilid-core/src/crypto/byte_array_types.rs
View File

@ -217,6 +217,14 @@ macro_rules! byte_array_type {
}
}
impl FromStr for $name {
type Err = VeilidAPIError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
$name::try_from(s)
}
}
impl TryFrom<String> for $name {
type Error = VeilidAPIError;
fn try_from(value: String) -> Result<Self, Self::Error> {

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

@ -634,7 +634,7 @@ impl NetworkManager {
pub async fn handle_private_receipt<R: AsRef<[u8]>>(
&self,
receipt_data: R,
private_route: TypedKey,
private_route: PublicKey,
) -> NetworkResult<()> {
let receipt_manager = self.receipt_manager();

4
veilid-core/src/receipt_manager.rs
View File

@ -11,7 +11,7 @@ pub enum ReceiptEvent {
ReturnedOutOfBand,
ReturnedInBand { inbound_noderef: NodeRef },
ReturnedSafety,
ReturnedPrivate { private_route: TypedKey },
ReturnedPrivate { private_route: PublicKey },
Expired,
Cancelled,
}
@ -21,7 +21,7 @@ pub enum ReceiptReturned {
OutOfBand,
InBand { inbound_noderef: NodeRef },
Safety,
Private { private_route: TypedKey },
Private { private_route: PublicKey },
}
pub trait ReceiptCallback: Send + 'static {

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

@ -891,11 +891,20 @@ impl RoutingTable {
}
#[instrument(level = "trace", skip(self), ret)]
pub fn register_find_node_answer(&self, peers: Vec<PeerInfo>) -> Vec<NodeRef> {
// register nodes we'd found
pub fn register_find_node_answer(
&self,
crypto_kind: CryptoKind,
peers: Vec<PeerInfo>,
) -> Vec<NodeRef> {
// Register nodes we'd found
let mut out = Vec::<NodeRef>::with_capacity(peers.len());
for p in peers {
// register the node if it's new
// Ensure we're getting back nodes we asked for
if !p.node_ids.kinds().contains(&crypto_kind) {
continue;
}
// Register the node if it's new
if let Some(nr) =
self.register_node_with_peer_info(RoutingDomain::PublicInternet, p, false)
{
@ -922,29 +931,41 @@ impl RoutingTable {
// register nodes we'd found
Ok(NetworkResult::value(
self.register_find_node_answer(res.answer),
self.register_find_node_answer(node_id.kind, res.answer),
))
}
/// Ask a remote node to list the nodes it has around the current node
#[instrument(level = "trace", skip(self), ret, err)]
pub async fn find_self(&self, node_ref: NodeRef) -> EyreResult<NetworkResult<Vec<NodeRef>>> {
let self_node_id = self.node_id();
pub async fn find_self(
&self,
crypto_kind: CryptoKind,
node_ref: NodeRef,
) -> EyreResult<NetworkResult<Vec<NodeRef>>> {
let self_node_id = self.node_id(crypto_kind);
self.find_node(node_ref, self_node_id).await
}
/// Ask a remote node to list the nodes it has around itself
#[instrument(level = "trace", skip(self), ret, err)]
pub async fn find_target(&self, node_ref: NodeRef) -> EyreResult<NetworkResult<Vec<NodeRef>>> {
let target_node_id = node_ref.node_id();
pub async fn find_target(
&self,
crypto_kind: CryptoKind,
node_ref: NodeRef,
) -> EyreResult<NetworkResult<Vec<NodeRef>>> {
let Some(target_node_id) = node_ref.node_ids().get(crypto_kind) else {
bail!("no target node ids for this crypto kind");
};
self.find_node(node_ref, target_node_id).await
}
#[instrument(level = "trace", skip(self))]
pub async fn reverse_find_node(&self, node_ref: NodeRef, wide: bool) {
// Ask bootstrap node to 'find' our own node so we can get some more nodes near ourselves
pub async fn reverse_find_node(&self, crypto_kind: CryptoKind, node_ref: NodeRef, wide: bool) {
// Ask node to 'find node' on own node so we can get some more nodes near ourselves
// and then contact those nodes to inform -them- that we exist
// Ask bootstrap server for nodes closest to our own node
let closest_nodes = network_result_value_or_log!(match self.find_self(node_ref.clone()).await {
// Ask node for nodes closest to our own node
let closest_nodes = network_result_value_or_log!(match self.find_self(crypto_kind, node_ref.clone()).await {
Err(e) => {
log_rtab!(error
"find_self failed for {:?}: {:?}",
@ -960,7 +981,7 @@ impl RoutingTable {
// Ask each node near us to find us as well
if wide {
for closest_nr in closest_nodes {
network_result_value_or_log!(match self.find_self(closest_nr.clone()).await {
network_result_value_or_log!(match self.find_self(crypto_kind, closest_nr.clone()).await {
Err(e) => {
log_rtab!(error
"find_self failed for {:?}: {:?}",

112
veilid-core/src/routing_table/route_spec_store.rs
View File

@ -168,7 +168,7 @@ pub struct RouteSpecDetail {
/// Secret key
#[with(Skip)]
secret_key: SecretKey,
/// Route hops
/// Route hops (node id keys)
hops: Vec<PublicKey>,
/// Route noderefs
#[with(Skip)]
@ -444,7 +444,7 @@ impl RouteSpecStore {
let mut dead_keys = Vec::new();
for (k, rsd) in &mut content.details {
for h in &rsd.hops {
let Some(nr) = routing_table.lookup_node_ref(*h) else {
let Some(nr) = routing_table.lookup_node_ref(TypedKey::new(rsd.crypto_kind, *h)) else {
dead_keys.push(*k);
break;
};
@ -582,13 +582,13 @@ impl RouteSpecStore {
fn detail<'a>(
inner: &'a RouteSpecStoreInner,
route_spec_key: &TypedKey,
route_spec_key: &PublicKey,
) -> Option<&'a RouteSpecDetail> {
inner.content.details.get(route_spec_key)
}
fn detail_mut<'a>(
inner: &'a mut RouteSpecStoreInner,
route_spec_key: &TypedKey,
route_spec_key: &PublicKey,
) -> Option<&'a mut RouteSpecDetail> {
inner.content.details.get_mut(route_spec_key)
}
@ -1594,9 +1594,10 @@ impl RouteSpecStore {
&self,
inner: &mut RouteSpecStoreInner,
rti: &RoutingTableInner,
crypto_kind: CryptoKind,
safety_spec: &SafetySpec,
direction: DirectionSet,
avoid_nodes: &[PublicKey],
avoid_nodes: &[TypedKey],
) -> EyreResult<Option<PublicKey>> {
// Ensure the total hop count isn't too long for our config
let max_route_hop_count = self.unlocked_inner.max_route_hop_count;
@ -1610,12 +1611,15 @@ impl RouteSpecStore {
// See if the preferred route is here
if let Some(preferred_route) = safety_spec.preferred_route {
if let Some(preferred_rsd) = inner.content.details.get(&preferred_route) {
// Only use the preferred route if it has the desire crypto kind
if preferred_rsd.crypto_kind == crypto_kind {
// Only use the preferred route if it doesn't end with the avoid nodes
if !avoid_node_ids.contains(preferred_rsd.hops.last().unwrap()) {
if !avoid_nodes.contains(&TypedKey::new(crypto_kind, preferred_rsd.hops.last().cloned().unwrap())) {
return Ok(Some(preferred_route));
}
}
}
}
// Select a safety route from the pool or make one if we don't have one that matches
let sr_pubkey = if let Some(sr_pubkey) = Self::first_available_route_inner(
@ -1657,7 +1661,7 @@ impl RouteSpecStore {
&self,
crypto_kind: CryptoKind,
safety_spec: &SafetySpec,
avoid_nodes: &[PublicKey],
avoid_nodes: &[TypedKey],
) -> EyreResult<Option<PublicKey>> {
let inner = &mut *self.inner.lock();
let routing_table = self.unlocked_inner.routing_table.clone();
@ -1666,6 +1670,7 @@ impl RouteSpecStore {
Ok(self.get_route_for_safety_spec_inner(
inner,
rti,
crypto_kind,
safety_spec,
Direction::Inbound.into(),
avoid_nodes,
@ -1673,6 +1678,7 @@ impl RouteSpecStore {
}
/// Assemble private route for publication
/// Returns a PrivateRoute object for an allocated private route
#[instrument(level = "trace", skip(self), err)]
pub fn assemble_private_route(
&self,
@ -1683,8 +1689,15 @@ impl RouteSpecStore {
let routing_table = self.unlocked_inner.routing_table.clone();
let rti = &*routing_table.inner.read();
// Get the route spec detail for the requested private route
let rsd = Self::detail(inner, key).ok_or_else(|| eyre!("route does not exist"))?;
// Ensure we get the crypto for it
let crypto = routing_table.network_manager().crypto();
let Some(vcrypto) = crypto.get(rsd.crypto_kind) else {
bail!("crypto not supported for route");
};
// See if we can optimize this compilation yet
// We don't want to include full nodeinfo if we don't have to
let optimized = optimized
@ -1696,7 +1709,10 @@ impl RouteSpecStore {
if !rti.has_valid_own_node_info(RoutingDomain::PublicInternet) {
bail!("can't make private routes until our node info is valid");
}
RouteNode::NodeId(NodeId::new(routing_table.node_id()))
let Some(node_id) = routing_table.node_ids().get(rsd.crypto_kind) else {
bail!("missing node id for crypto kind");
};
RouteNode::NodeId(node_id.key)
} else {
let Some(pi) = rti.get_own_peer_info(RoutingDomain::PublicInternet) else {
bail!("can't make private routes until our node info is valid");
@ -1706,12 +1722,11 @@ impl RouteSpecStore {
next_hop: None,
};
let crypto = routing_table.network_manager().crypto();
// Loop for each hop
let hop_count = rsd.hops.len();
// iterate hops in private route order (reverse, but inside out)
for h in 0..hop_count {
let nonce = Crypto::get_random_nonce();
let nonce = vcrypto.random_nonce();
let blob_data = {
let mut rh_message = ::capnp::message::Builder::new_default();
@ -1721,10 +1736,10 @@ impl RouteSpecStore {
};
// Encrypt the previous blob ENC(nonce, DH(PKhop,SKpr))
let dh_secret = crypto
let dh_secret = vcrypto
.cached_dh(&rsd.hops[h], &rsd.secret_key)
.wrap_err("dh failed")?;
let enc_msg_data = Crypto::encrypt_aead(blob_data.as_slice(), &nonce, &dh_secret, None)
let enc_msg_data = vcrypto.encrypt_aead(blob_data.as_slice(), &nonce, &dh_secret, None)
.wrap_err("encryption failed")?;
let route_hop_data = RouteHopData {
nonce,
@ -1734,14 +1749,14 @@ impl RouteSpecStore {
route_hop = RouteHop {
node: if optimized {
// Optimized, no peer info, just the dht key
RouteNode::NodeId(NodeId::new(rsd.hops[h]))
RouteNode::NodeId(rsd.hops[h])
} else {
// Full peer info, required until we are sure the route has been fully established
let node_id = rsd.hops[h];
let node_id = TypedKey::new(rsd.crypto_kind, rsd.hops[h]);
let pi = rti
.with_node_entry(node_id, |entry| {
entry.with(rti, |_rti, e| {
e.make_peer_info(node_id, RoutingDomain::PublicInternet)
e.make_peer_info(RoutingDomain::PublicInternet)
})
})
.flatten();
@ -1755,7 +1770,7 @@ impl RouteSpecStore {
}
let private_route = PrivateRoute {
public_key: key.clone(),
public_key: TypedKey::new(rsd.crypto_kind, key.clone()),
// add hop for 'FirstHop'
hop_count: (hop_count + 1).try_into().unwrap(),
hops: PrivateRouteHops::FirstHop(route_hop),
@ -1765,9 +1780,14 @@ impl RouteSpecStore {
/// Import a remote private route for compilation
#[instrument(level = "trace", skip(self, blob), ret, err)]
pub fn import_remote_private_route(&self, blob: Vec<u8>) -> EyreResult<PublicKey> {
pub fn import_remote_private_route(&self, blob: Vec<u8>) -> EyreResult<TypedKeySet> {
// decode the pr blob
let private_route = RouteSpecStore::blob_to_private_route(blob)?;
let private_routes = RouteSpecStore::blob_to_private_routes(blob)?;
let mut out = TypedKeySet::new();
let inner = &mut *self.inner.lock();
for private_route in private_routes {
// ensure private route has first hop
if !matches!(private_route.hops, PrivateRouteHops::FirstHop(_)) {
@ -1775,8 +1795,7 @@ impl RouteSpecStore {
}
// ensure this isn't also an allocated route
let inner = &mut *self.inner.lock();
if Self::detail(inner, &private_route.public_key).is_some() {
if Self::detail(inner, &private_route.public_key.key).is_some() {
bail!("should not import allocated route");
}
@ -1785,7 +1804,11 @@ impl RouteSpecStore {
let key = Self::with_create_remote_private_route(inner, cur_ts, private_route, |r| {
r.private_route.as_ref().unwrap().public_key.clone()
});
Ok(key)
out.add(key);
}
Ok(out)
}
/// Release a remote private route that is no longer in use
@ -1829,7 +1852,7 @@ impl RouteSpecStore {
where
F: FnOnce(&mut RemotePrivateRouteInfo) -> R,
{
let pr_pubkey = private_route.public_key;
let pr_pubkey = private_route.public_key.key;
let rpr = inner
.cache
@ -2045,24 +2068,52 @@ impl RouteSpecStore {
}
}
/// Convert private route to binary blob
pub fn private_route_to_blob(private_route: &PrivateRoute) -> EyreResult<Vec<u8>> {
/// Convert private route list to binary blob
pub fn private_routes_to_blob(private_routes: &[PrivateRoute]) -> EyreResult<Vec<u8>> {
let mut pr_message = ::capnp::message::Builder::new_default();
let mut pr_builder = pr_message.init_root::<veilid_capnp::private_route::Builder>();
encode_private_route(&private_route, &mut pr_builder)
.wrap_err("failed to encode private route")?;
let mut buffer = vec![];
// Serialize count
let pr_count = private_routes.len();
if pr_count > MAX_CRYPTO_KINDS {
bail!("too many crypto kinds to encode blob");
}
let pr_count = pr_count as u8;
buffer.push(pr_count);
// Serialize stream of private routes
for private_route in private_routes {
encode_private_route(private_route, &mut pr_builder)
.wrap_err("failed to encode private route")?;
capnp::serialize_packed::write_message(&mut buffer, &pr_message)
.map_err(RPCError::internal)
.wrap_err("failed to convert builder to vec")?;
}
Ok(buffer)
}
/// Convert binary blob to private route
pub fn blob_to_private_route(blob: Vec<u8>) -> EyreResult<PrivateRoute> {
pub fn blob_to_private_routes(blob: Vec<u8>) -> EyreResult<Vec<PrivateRoute>> {
// Deserialize count
if blob.is_empty() {
bail!("not deserializing empty private route blob");
}
let pr_count = blob[0] as usize;
if pr_count > MAX_CRYPTO_KINDS {
bail!("too many crypto kinds to decode blob");
}
// Deserialize stream of private routes
let pr_slice = &blob[1..];
let mut out = Vec::with_capacity(pr_count);
for _ in 0..pr_count {
let reader = capnp::serialize_packed::read_message(
blob.as_slice(),
&mut pr_slice,
capnp::message::ReaderOptions::new(),
)
.map_err(RPCError::internal)
@ -2072,6 +2123,9 @@ impl RouteSpecStore {
.get_root::<veilid_capnp::private_route::Reader>()
.map_err(RPCError::internal)
.wrap_err("failed to make reader for private_route")?;
decode_private_route(&pr_reader).wrap_err("failed to decode private route")
let private_route = decode_private_route(&pr_reader).wrap_err("failed to decode private route")?;
out.push(private_route);
}
Ok(out)
}
}

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

@ -413,6 +413,7 @@ impl RoutingTableInner {
&self,
routing_domain_set: RoutingDomainSet,
min_state: BucketEntryState,
crypto_kinds: &[CryptoKind], xxx finish this and peer minimum refresh and bootstrap tick, then both routines
) -> usize {
let mut count = 0usize;
let cur_ts = get_aligned_timestamp();
@ -451,7 +452,7 @@ impl RoutingTableInner {
mut f: F,
) -> Option<T> {
for entry in self.all_entries {
if entry.with(self, |_rti, e| e.state(cur_ts) >= min_state) {
if entry.with_inner(|e| e.state(cur_ts) >= min_state) {
if let Some(out) = f(self, entry) {
return Some(out);
}
@ -474,7 +475,7 @@ impl RoutingTableInner {
// Collect all entries that are 'needs_ping' and have some node info making them reachable somehow
let mut node_refs = Vec::<NodeRef>::with_capacity(self.bucket_entry_count());
self.with_entries(cur_ts, BucketEntryState::Unreliable, |rti, entry| {
if entry.with(rti, |rti, e| {
if entry.with_inner(|e| {
// If this isn't in the routing domain we are checking, don't include it
if !e.exists_in_routing_domain(rti, routing_domain) {
return false;
@ -757,7 +758,7 @@ impl RoutingTableInner {
let cur_ts = get_aligned_timestamp();
for entry in self.all_entries {
match entry.with(self, |_rti, e| e.state(cur_ts)) {
match entry.with_inner(|e| e.state(cur_ts)) {
BucketEntryState::Reliable => {
reliable_entry_count += 1;
}
@ -807,7 +808,7 @@ impl RoutingTableInner {
) -> Vec<NodeRef> {
let public_node_filter = Box::new(|rti: &RoutingTableInner, v: Option<Arc<BucketEntry>>| {
let entry = v.unwrap();
entry.with(rti, |_rti, e| {
entry.with_inner(|e| {
// skip nodes on local network
if e.node_info(RoutingDomain::LocalNetwork).is_some() {
return false;
@ -838,7 +839,7 @@ impl RoutingTableInner {
) -> bool {
match entry {
None => has_valid_own_node_info,
Some(entry) => entry.with(self, |_rti, e| {
Some(entry) => entry.with_inner(|e| {
e.signed_node_info(routing_domain.into())
.map(|sni| sni.has_any_signature())
.unwrap_or(false)
@ -854,7 +855,7 @@ impl RoutingTableInner {
) -> PeerInfo {
match entry {
None => own_peer_info.clone(),
Some(entry) => entry.with(self, |_rti, e| e.make_peer_info(routing_domain).unwrap()),
Some(entry) => entry.with_inner(|e| e.make_peer_info(routing_domain).unwrap()),
}
}
@ -932,7 +933,7 @@ impl RoutingTableInner {
move |rti: &RoutingTableInner, v: Option<Arc<BucketEntry>>| {
if let Some(entry) = &v {
// always filter out dead nodes
if entry.with(rti, |_rti, e| e.state(cur_ts) == BucketEntryState::Dead) {
if entry.with_inner(|e| e.state(cur_ts) == BucketEntryState::Dead) {
false
} else {
true
@ -970,8 +971,8 @@ impl RoutingTableInner {
// reliable nodes come first
let ae = a_entry.as_ref().unwrap();
let be = b_entry.as_ref().unwrap();
ae.with(rti, |rti, ae| {
be.with(rti, |_rti, be| {
ae.with_inner(|ae| {
be.with_inner(|be| {
let ra = ae.check_reliable(cur_ts);
let rb = be.check_reliable(cur_ts);
if ra != rb {
@ -1020,7 +1021,25 @@ impl RoutingTableInner {
{
let cur_ts = get_aligned_timestamp();
// closest sort
// Get the crypto kind
let crypto_kind = node_id.kind;
let vcrypto = self.unlocked_inner.crypto().get(crypto_kind).unwrap();
// Filter to ensure entries support the crypto kind in use
let filter = Box::new(
move |rti: &RoutingTableInner, opt_entry: Option<Arc<BucketEntry>>| {
if let Some(entry) = opt_entry {
entry.with_inner(|e| e.crypto_kinds().contains(&crypto_kind))
} else {
VALID_CRYPTO_KINDS.contains(&crypto_kind)
}
},
) as RoutingTableEntryFilter;
filters.push_front(filter);
// Closest sort
// Distance is done using the node id's distance metric which may vary based on crypto system
let sort = |rti: &RoutingTableInner,
a_entry: &Option<Arc<BucketEntry>>,
b_entry: &Option<Arc<BucketEntry>>| {
@ -1038,10 +1057,10 @@ impl RoutingTableInner {
// reliable nodes come first, pessimistically treating our own node as unreliable
let ra = a_entry
.as_ref()
.map_or(false, |x| x.with(rti, |_rti, x| x.check_reliable(cur_ts)));
.map_or(false, |x| x.with_inner(|x| x.check_reliable(cur_ts)));
let rb = b_entry
.as_ref()
.map_or(false, |x| x.with(rti, |_rti, x| x.check_reliable(cur_ts)));
.map_or(false, |x| x.with_inner(|x| x.check_reliable(cur_ts)));
if ra != rb {
if ra {
return core::cmp::Ordering::Less;
@ -1050,9 +1069,24 @@ impl RoutingTableInner {
}
}
// get keys
let a_key = if let Some(a_entry) = a_entry {
a_entry.with_inner(|e| e.node_ids().get(crypto_kind).unwrap())
} else {
self.unlocked_inner.node_id(crypto_kind)
};
let b_key = if let Some(b_entry) = b_entry {
b_entry.with_inner(|e| e.node_ids().get(crypto_kind).unwrap())
} else {
self.unlocked_inner.node_id(crypto_kind)
};
// distance is the next metric, closer nodes first
let da = distance(a_key, &node_id);
let db = distance(b_key, &node_id);
// since multiple cryptosystems are in use, the distance for a key is the shortest
// distance to that key over all supported cryptosystems
let da = vcrypto.distance(&a_key.key, &node_id.key);
let db = vcrypto.distance(&b_key.key, &node_id.key);
da.cmp(&db)
};

16
veilid-core/src/routing_table/tasks/bootstrap.rs
View File

@ -262,15 +262,18 @@ impl RoutingTable {
self.register_node_with_peer_info(RoutingDomain::PublicInternet, pi, false)
{
// Add this our futures to process in parallel
for crypto_kind in VALID_CRYPTO_KINDS {
let routing_table = self.clone();
let nr = nr.clone();
unord.push(
// lets ask bootstrap to find ourselves now
async move { routing_table.reverse_find_node(nr, true).await }
async move { routing_table.reverse_find_node(crypto_kind, nr, true).await }
.instrument(Span::current()),
);
}
}
}
}
// Wait for all bootstrap operations to complete before we complete the singlefuture
while let Ok(Some(_)) = unord.next().timeout_at(stop_token.clone()).await {}
@ -337,15 +340,19 @@ impl RoutingTable {
self.register_node_with_peer_info(RoutingDomain::PublicInternet, pi, true)
{
// Add this our futures to process in parallel
for crypto_kind in VALID_CRYPTO_KINDS {
let nr = nr.clone();
let routing_table = self.clone();
unord.push(
async move {
// Need VALID signed peer info, so ask bootstrap to find_node of itself
// which will ensure it has the bootstrap's signed peer info as part of the response
let _ = routing_table.find_target(nr.clone()).await;
let _ = routing_table.find_target(crypto_kind, nr.clone()).await;
// Ensure we got the signed peer info
if !nr.signed_node_info_has_valid_signature(RoutingDomain::PublicInternet) {
if !nr
.signed_node_info_has_valid_signature(RoutingDomain::PublicInternet)
{
log_rtab!(warn
"bootstrap at {:?} did not return valid signed node info",
nr
@ -353,13 +360,14 @@ impl RoutingTable {
// If this node info is invalid, it will time out after being unpingable
} else {
// otherwise this bootstrap is valid, lets ask it to find ourselves now
routing_table.reverse_find_node(nr, true).await
routing_table.reverse_find_node(crypto_kind, nr, true).await
}
}
.instrument(Span::current()),
);
}
}
}
// Wait for all bootstrap operations to complete before we complete the singlefuture
while let Ok(Some(_)) = unord.next().timeout_at(stop_token.clone()).await {}

21
veilid-core/src/routing_table/tasks/peer_minimum_refresh.rs
View File

@ -20,16 +20,32 @@ impl RoutingTable {
// For the PublicInternet routing domain, get list of all peers we know about
// even the unreliable ones, and ask them to find nodes close to our node too
let mut ord = FuturesOrdered::new();
for crypto_kind in VALID_CRYPTO_KINDS {
let routing_table = self.clone();
let mut filters = VecDeque::new();
let filter = Box::new(
move |rti: &RoutingTableInner, opt_entry: Option<Arc<BucketEntry>>| {
if let Some(entry) = opt_entry {
entry.with_inner(|e| e.crypto_kinds().contains(&crypto_kind))
} else {
VALID_CRYPTO_KINDS.contains(&crypto_kind)
}
},
) as RoutingTableEntryFilter;
filters.push_front(filter);
let noderefs = routing_table.find_fastest_nodes(
min_peer_count,
VecDeque::new(),
filters,
|_rti, entry: Option<Arc<BucketEntry>>| {
NodeRef::new(routing_table.clone(), entry.unwrap().clone(), None)
},
);
let mut ord = FuturesOrdered::new();
for nr in noderefs {
let routing_table = self.clone();
ord.push_back(
@ -37,6 +53,7 @@ impl RoutingTable {
.instrument(Span::current()),
);
}
}
// do peer minimum search in order from fastest to slowest
while let Ok(Some(_)) = ord.next().timeout_at(stop_token.clone()).await {}

12
veilid-core/src/rpc_processor/coders/operations/answer.rs
View File

@ -15,9 +15,12 @@ impl RPCAnswer {
pub fn desc(&self) -> &'static str {
self.detail.desc()
}
pub fn decode(reader: &veilid_capnp::answer::Reader) -> Result<RPCAnswer, RPCError> {
pub fn decode(
reader: &veilid_capnp::answer::Reader,
crypto: Crypto,
) -> Result<RPCAnswer, RPCError> {
let d_reader = reader.get_detail();
let detail = RPCAnswerDetail::decode(&d_reader)?;
let detail = RPCAnswerDetail::decode(&d_reader, crypto)?;
Ok(RPCAnswer { detail })
}
pub fn encode(&self, builder: &mut veilid_capnp::answer::Builder) -> Result<(), RPCError> {
@ -60,6 +63,7 @@ impl RPCAnswerDetail {
pub fn decode(
reader: &veilid_capnp::answer::detail::Reader,
crypto: Crypto,
) -> Result<RPCAnswerDetail, RPCError> {
let which_reader = reader.which().map_err(RPCError::protocol)?;
let out = match which_reader {
@ -70,7 +74,7 @@ impl RPCAnswerDetail {
}
veilid_capnp::answer::detail::FindNodeA(r) => {
let op_reader = r.map_err(RPCError::protocol)?;
let out = RPCOperationFindNodeA::decode(&op_reader)?;
let out = RPCOperationFindNodeA::decode(&op_reader, crypto)?;
RPCAnswerDetail::FindNodeA(out)
}
veilid_capnp::answer::detail::AppCallA(r) => {
@ -100,7 +104,7 @@ impl RPCAnswerDetail {
}
veilid_capnp::answer::detail::FindBlockA(r) => {
let op_reader = r.map_err(RPCError::protocol)?;
let out = RPCOperationFindBlockA::decode(&op_reader)?;
let out = RPCOperationFindBlockA::decode(&op_reader, crypto)?;
RPCAnswerDetail::FindBlockA(out)
}
veilid_capnp::answer::detail::StartTunnelA(r) => {

4
veilid-core/src/rpc_processor/coders/operations/operation.rs
View File

@ -29,12 +29,12 @@ impl RPCOperationKind {
}
veilid_capnp::operation::kind::Which::Statement(r) => {
let q_reader = r.map_err(RPCError::protocol)?;
let out = RPCStatement::decode(&q_reader)?;
let out = RPCStatement::decode(&q_reader, crypto)?;
RPCOperationKind::Statement(out)
}
veilid_capnp::operation::kind::Which::Answer(r) => {
let q_reader = r.map_err(RPCError::protocol)?;
let out = RPCAnswer::decode(&q_reader)?;
let out = RPCAnswer::decode(&q_reader, crypto)?;
RPCOperationKind::Answer(out)
}
};

5
veilid-core/src/rpc_processor/coders/operations/operation_find_block.rs
View File

@ -35,6 +35,7 @@ pub struct RPCOperationFindBlockA {
impl RPCOperationFindBlockA {
pub fn decode(
reader: &veilid_capnp::operation_find_block_a::Reader,
crypto: Crypto,
) -> Result<RPCOperationFindBlockA, RPCError> {
let data = reader.get_data().map_err(RPCError::protocol)?.to_vec();
@ -46,7 +47,7 @@ impl RPCOperationFindBlockA {
.map_err(RPCError::map_internal("too many suppliers"))?,
);
for s in suppliers_reader.iter() {
let peer_info = decode_peer_info(&s)?;
let peer_info = decode_peer_info(&s, crypto.clone())?;
suppliers.push(peer_info);
}
@ -58,7 +59,7 @@ impl RPCOperationFindBlockA {
.map_err(RPCError::map_internal("too many peers"))?,
);
for p in peers_reader.iter() {
let peer_info = decode_peer_info(&p)?;
let peer_info = decode_peer_info(&p, crypto.clone())?;
peers.push(peer_info);
}

7
veilid-core/src/rpc_processor/coders/operations/operation_find_node.rs
View File

@ -10,7 +10,7 @@ impl RPCOperationFindNodeQ {
reader: &veilid_capnp::operation_find_node_q::Reader,
) -> Result<RPCOperationFindNodeQ, RPCError> {
let ni_reader = reader.get_node_id().map_err(RPCError::protocol)?;
let node_id = decode_key256(&ni_reader);
let node_id = decode_typed_key(&ni_reader)?;
Ok(RPCOperationFindNodeQ { node_id })
}
pub fn encode(
@ -18,7 +18,7 @@ impl RPCOperationFindNodeQ {
builder: &mut veilid_capnp::operation_find_node_q::Builder,
) -> Result<(), RPCError> {
let mut ni_builder = builder.reborrow().init_node_id();
encode_key256(&self.node_id, &mut ni_builder)?;
encode_typed_key(&self.node_id, &mut ni_builder);
Ok(())
}
}
@ -31,6 +31,7 @@ pub struct RPCOperationFindNodeA {
impl RPCOperationFindNodeA {
pub fn decode(
reader: &veilid_capnp::operation_find_node_a::Reader,
crypto: Crypto,
) -> Result<RPCOperationFindNodeA, RPCError> {
let peers_reader = reader.get_peers().map_err(RPCError::protocol)?;
let mut peers = Vec::<PeerInfo>::with_capacity(
@ -40,7 +41,7 @@ impl RPCOperationFindNodeA {
.map_err(RPCError::map_internal("too many peers"))?,
);
for p in peers_reader.iter() {
let peer_info = decode_peer_info(&p)?;
let peer_info = decode_peer_info(&p, crypto.clone())?;
peers.push(peer_info);
}

5
veilid-core/src/rpc_processor/coders/operations/operation_route.rs
View File

@ -29,7 +29,7 @@ impl RoutedOperation {
.map_err(RPCError::map_internal("too many signatures"))?,
);
for s in sigs_reader.iter() {
let sig = decode_typed_signature(&s);
let sig = decode_typed_signature(&s)?;
signatures.push(sig);
}
@ -80,9 +80,10 @@ pub struct RPCOperationRoute {
impl RPCOperationRoute {
pub fn decode(
reader: &veilid_capnp::operation_route::Reader,
crypto: Crypto,
) -> Result<RPCOperationRoute, RPCError> {
let sr_reader = reader.get_safety_route().map_err(RPCError::protocol)?;
let safety_route = decode_safety_route(&sr_reader)?;
let safety_route = decode_safety_route(&sr_reader, crypto)?;
let o_reader = reader.get_operation().map_err(RPCError::protocol)?;
let operation = RoutedOperation::decode(&o_reader)?;

3
veilid-core/src/rpc_processor/coders/operations/operation_signal.rs
View File

@ -8,8 +8,9 @@ pub struct RPCOperationSignal {
impl RPCOperationSignal {
pub fn decode(
reader: &veilid_capnp::operation_signal::Reader,
crypto: Crypto,
) -> Result<RPCOperationSignal, RPCError> {
let signal_info = decode_signal_info(reader)?;
let signal_info = decode_signal_info(reader, crypto)?;
Ok(RPCOperationSignal { signal_info })
}
pub fn encode(

12
veilid-core/src/rpc_processor/coders/operations/statement.rs
View File

@ -18,9 +18,12 @@ impl RPCStatement {
pub fn desc(&self) -> &'static str {
self.detail.desc()
}
pub fn decode(reader: &veilid_capnp::statement::Reader) -> Result<RPCStatement, RPCError> {
pub fn decode(
reader: &veilid_capnp::statement::Reader,
crypto: Crypto,
) -> Result<RPCStatement, RPCError> {
let d_reader = reader.get_detail();
let detail = RPCStatementDetail::decode(&d_reader)?;
let detail = RPCStatementDetail::decode(&d_reader, crypto)?;
Ok(RPCStatement { detail })
}
pub fn encode(&self, builder: &mut veilid_capnp::statement::Builder) -> Result<(), RPCError> {
@ -52,6 +55,7 @@ impl RPCStatementDetail {
}
pub fn decode(
reader: &veilid_capnp::statement::detail::Reader,
crypto: Crypto,
) -> Result<RPCStatementDetail, RPCError> {
let which_reader = reader.which().map_err(RPCError::protocol)?;
let out = match which_reader {
@ -62,7 +66,7 @@ impl RPCStatementDetail {
}
veilid_capnp::statement::detail::Route(r) => {
let op_reader = r.map_err(RPCError::protocol)?;
let out = RPCOperationRoute::decode(&op_reader)?;
let out = RPCOperationRoute::decode(&op_reader, crypto)?;
RPCStatementDetail::Route(out)
}
veilid_capnp::statement::detail::ValueChanged(r) => {
@ -72,7 +76,7 @@ impl RPCStatementDetail {
}
veilid_capnp::statement::detail::Signal(r) => {
let op_reader = r.map_err(RPCError::protocol)?;
let out = RPCOperationSignal::decode(&op_reader)?;
let out = RPCOperationSignal::decode(&op_reader, crypto)?;
RPCStatementDetail::Signal(out)
}
veilid_capnp::statement::detail::ReturnReceipt(r) => {

4
veilid-core/src/rpc_processor/destination.rs
View File

@ -164,7 +164,7 @@ impl RPCProcessor {
// Sent directly but with a safety route, respond to private route
let ck = target.best_node_id().kind;
let Some(pr_key) = rss
.get_private_route_for_safety_spec(ck, safety_spec, &target.node_ids().keys())
.get_private_route_for_safety_spec(ck, safety_spec, &target.node_ids())
.map_err(RPCError::internal)? else {
return Ok(NetworkResult::no_connection_other("no private route for response at this time"));
};
@ -193,7 +193,7 @@ impl RPCProcessor {
let mut avoid_nodes = relay.node_ids();
avoid_nodes.add_all(&target.node_ids());
let Some(pr_key) = rss
.get_private_route_for_safety_spec(ck, safety_spec, &avoid_nodes.keys())
.get_private_route_for_safety_spec(ck, safety_spec, &avoid_nodes)
.map_err(RPCError::internal)? else {
return Ok(NetworkResult::no_connection_other("no private route for response at this time"));
};

11
veilid-core/src/rpc_processor/mod.rs
View File

@ -87,7 +87,16 @@ struct RPCMessageHeader {
detail: RPCMessageHeaderDetail,
}
impl RPCMessageHeader {}
impl RPCMessageHeader {
/// The crypto kind used on the RPC
pub fn crypto_kind(&self) -> CryptoKind {
match &self.detail {
RPCMessageHeaderDetail::Direct(d) => d.envelope.get_crypto_kind(),
RPCMessageHeaderDetail::SafetyRouted(s) => todo!(),
RPCMessageHeaderDetail::PrivateRouted(p) => todo!(),
}
}
}
#[derive(Debug)]
pub struct RPCMessageData {

5
veilid-core/src/rpc_processor/rpc_app_call.rs
View File

@ -44,6 +44,9 @@ impl RPCProcessor {
&self,
msg: RPCMessage,
) -> Result<NetworkResult<()>, RPCError> {
// Get the crypto kind used to send this question
let crypto_kind = msg.header.crypto_kind();
// Get the question
let app_call_q = match msg.operation.kind() {
RPCOperationKind::Question(q) => match q.detail() {
@ -61,7 +64,7 @@ impl RPCProcessor {
let sender = msg
.opt_sender_nr
.as_ref()
.map(|nr| NodeId::new(nr.node_id()));
.map(|nr| nr.node_ids().get(crypto_kind).unwrap().key);
let message = app_call_q.message.clone();
(self.unlocked_inner.update_callback)(VeilidUpdate::AppCall(VeilidAppCall {
sender,

10
veilid-core/src/rpc_processor/rpc_app_message.rs
View File

@ -30,8 +30,16 @@ impl RPCProcessor {
_ => panic!("not a statement"),
};
// Get the crypto kind used to send this question
let crypto_kind = msg.header.crypto_kind();
// Get the sender node id this came from
let sender = msg
.opt_sender_nr
.as_ref()
.map(|nr| nr.node_ids().get(crypto_kind).unwrap().key);
// Pass the message up through the update callback
let sender = msg.opt_sender_nr.map(|nr| NodeId::new(nr.node_id()));
let message = app_message.message;
(self.unlocked_inner.update_callback)(VeilidUpdate::AppMessage(VeilidAppMessage {
sender,

28
veilid-core/src/rpc_processor/rpc_find_node.rs
View File

@ -11,7 +11,7 @@ impl RPCProcessor {
pub async fn rpc_call_find_node(
self,
dest: Destination,
key: PublicKey,
node_id: TypedKey,
) -> Result<NetworkResult<Answer<Vec<PeerInfo>>>, RPCError> {
// Ensure destination never has a private route
if matches!(
@ -26,8 +26,7 @@ impl RPCProcessor {
));
}
let find_node_q_detail =
RPCQuestionDetail::FindNodeQ(RPCOperationFindNodeQ { node_id: key });
let find_node_q_detail = RPCQuestionDetail::FindNodeQ(RPCOperationFindNodeQ { node_id });
let find_node_q = RPCQuestion::new(
network_result_try!(self.get_destination_respond_to(&dest)?),
find_node_q_detail,
@ -90,6 +89,13 @@ impl RPCProcessor {
_ => panic!("not a question"),
};
// Get the crypto kinds the requesting node is capable of
let crypto_kinds = if let Some(sender_nr) = msg.opt_sender_nr {
sender_nr.node_ids().kinds()
} else {
vec![msg.header.crypto_kind()]
};
// add node information for the requesting node to our routing table
let routing_table = self.routing_table();
let Some(own_peer_info) = routing_table.get_own_peer_info(RoutingDomain::PublicInternet) else {
@ -98,10 +104,20 @@ impl RPCProcessor {
};
// find N nodes closest to the target node in our routing table
let filter = Box::new(
move |rti: &RoutingTableInner, entry: Option<Arc<BucketEntry>>| {
rti.filter_has_valid_signed_node_info(RoutingDomain::PublicInternet, true, entry)
move |rti: &RoutingTableInner, opt_entry: Option<Arc<BucketEntry>>| {
// ensure the returned nodes have at least the crypto kind used to send the findnodeq
if let Some(entry) = opt_entry {
if !entry.with(rti, |_rti, e| e.crypto_kinds().contains(&crypto_kind)) {
return false;
}
}
// Ensure only things that are valid/signed in the PublicInternet domain are returned
rti.filter_has_valid_signed_node_info(
RoutingDomain::PublicInternet,
true,
opt_entry,
)
},
) as RoutingTableEntryFilter;
let filters = VecDeque::from([filter]);

3
veilid-core/src/rpc_processor/rpc_status.rs
View File

@ -42,7 +42,6 @@ impl RPCProcessor {
target,
safety_selection: _,
} => {
let opt_target_nr = self.routing_table.lookup_node_ref(*target);
let routing_domain = match relay.best_routing_domain() {
Some(rd) => rd,
None => {
@ -51,7 +50,7 @@ impl RPCProcessor {
))
}
};
(opt_target_nr, routing_domain)
(Some(target.clone()), routing_domain)
}
Destination::PrivateRoute {
private_route: _,

9
veilid-core/src/rpc_processor/rpc_validate_dial_info.rs
View File

@ -88,7 +88,10 @@ impl RPCProcessor {
// Use the address type though, to ensure we reach an ipv6 capable node if this is
// an ipv6 address
let routing_table = self.routing_table();
let sender_id = detail.envelope.get_sender_id();
let sender_node_id = TypedKey::new(
detail.envelope.get_crypto_kind(),
detail.envelope.get_sender_id(),
);
let routing_domain = detail.routing_domain;
let node_count = {
let c = self.config.get();
@ -102,7 +105,7 @@ impl RPCProcessor {
dial_info.clone(),
);
let will_validate_dial_info_filter = Box::new(
move |rti: &RoutingTableInner, _k: TypedKey, v: Option<Arc<BucketEntry>>| {
move |rti: &RoutingTableInner, v: Option<Arc<BucketEntry>>| {
let entry = v.unwrap();
entry.with(rti, move |_rti, e| {
if let Some(status) = &e.node_status(routing_domain) {
@ -129,7 +132,7 @@ impl RPCProcessor {
}
for peer in peers {
// Ensure the peer is not the one asking for the validation
if peer.node_id() == sender_id {
if peer.node_ids().contains(&sender_node_id) {
continue;
}

14
veilid-core/src/veilid_api/api.rs
View File

@ -165,11 +165,10 @@ impl VeilidAPI {
// Private route allocation
/// Allocate a new private route set with default cryptography and network options
/// Returns a list of the public key and published 'blob' pairs. Publishing as many of these
/// pairs as possible to the network is desirable as support for multiple cryptography
/// systems will require choosing a compatible route
/// Returns a set of keys and a publishable 'blob' with the route encrypted with each crypto kind
/// Those nodes importing the blob will have their choice of which crypto kind to use
#[instrument(level = "debug", skip(self))]
pub async fn new_private_route(&self) -> Result<Vec<(PublicKey, Vec<u8>)>, VeilidAPIError> {
pub async fn new_private_route(&self) -> Result<(TypedKeySet, Vec<u8>), VeilidAPIError> {
self.new_custom_private_route(
&VALID_CRYPTO_KINDS,
Stability::default(),
@ -185,7 +184,7 @@ impl VeilidAPI {
crypto_kinds: &[CryptoKind],
stability: Stability,
sequencing: Sequencing,
) -> Result<Vec<(PublicKey, Vec<u8>)>, VeilidAPIError> {
) -> Result<(TypedKeySet, Vec<u8>), VeilidAPIError> {
let default_route_hop_count: usize = {
let config = self.config()?;
let c = config.get();
@ -232,7 +231,10 @@ impl VeilidAPI {
}
#[instrument(level = "debug", skip(self))]
pub fn import_remote_private_route(&self, blob: Vec<u8>) -> Result<PublicKey, VeilidAPIError> {
pub fn import_remote_private_route(
&self,
blob: Vec<u8>,
) -> Result<TypedKeySet, VeilidAPIError> {
let rss = self.routing_table()?.route_spec_store();
rss.import_remote_private_route(blob)
.map_err(|e| VeilidAPIError::invalid_argument(e, "blob", "private route blob"))

2
veilid-core/src/veilid_api/error.rs
View File

@ -108,7 +108,7 @@ pub enum VeilidAPIError {
#[error("Shutdown")]
Shutdown,
#[error("Key not found: {key}")]
KeyNotFound { key: TypedKey },
KeyNotFound { key: PublicKey },
#[error("No connection: {message}")]
NoConnection { message: String },
#[error("No peer info: {node_id}")]