diff --git a/doc/config/sample.config b/doc/config/sample.config index f4693374..88665c76 100644 --- a/doc/config/sample.config +++ b/doc/config/sample.config @@ -63,7 +63,7 @@ core: max_timestamp_behind_ms: 10000 max_timestamp_ahead_ms: 10000 timeout_ms: 10000 - max_route_hop_count: 5 + max_route_hop_count: 4 default_route_hop_count: 2 dht: diff --git a/veilid-core/proto/veilid.capnp b/veilid-core/proto/veilid.capnp index 0d1ec7e3..97904483 100644 --- a/veilid-core/proto/veilid.capnp +++ b/veilid-core/proto/veilid.capnp @@ -258,9 +258,10 @@ struct PeerInfo { } struct RoutedOperation { - signatures @0 :List(Signature); # signatures from nodes that have handled the private route - nonce @1 :Nonce; # nonce Xmsg - data @2 :Data; # Operation encrypted with ENC(Xmsg,DH(PKapr,SKbsr)) + version @0 :UInt8; # crypto version in use for the data + signatures @1 :List(Signature); # signatures from nodes that have handled the private route + nonce @2 :Nonce; # nonce Xmsg + data @3 :Data; # operation encrypted with ENC(Xmsg,DH(PKapr,SKbsr)) } struct OperationStatusQ { diff --git a/veilid-core/src/attachment_manager.rs b/veilid-core/src/attachment_manager.rs index 3d283140..0d834561 100644 --- a/veilid-core/src/attachment_manager.rs +++ b/veilid-core/src/attachment_manager.rs @@ -1,5 +1,5 @@ use crate::callback_state_machine::*; -use crate::dht::Crypto; +use crate::crypto::Crypto; use crate::network_manager::*; use crate::routing_table::*; use crate::xx::*; diff --git a/veilid-core/src/core_context.rs b/veilid-core/src/core_context.rs index 2640f455..698a6a84 100644 --- a/veilid-core/src/core_context.rs +++ b/veilid-core/src/core_context.rs @@ -1,6 +1,6 @@ use crate::api_tracing_layer::*; use crate::attachment_manager::*; -use crate::dht::Crypto; +use crate::crypto::Crypto; use crate::veilid_api::*; use crate::veilid_config::*; use crate::xx::*; @@ -103,7 +103,13 @@ impl ServicesContext { // Set up attachment manager trace!("init attachment manager"); let update_callback = self.update_callback.clone(); - let attachment_manager = AttachmentManager::new(self.config.clone(), protected_store, table_store, block_store, crypto); + let attachment_manager = AttachmentManager::new( + self.config.clone(), + protected_store, + table_store, + block_store, + crypto, + ); if let Err(e) = attachment_manager.init(update_callback).await { self.shutdown().await; return Err(e); diff --git a/veilid-core/src/dht/envelope.rs b/veilid-core/src/crypto/envelope.rs similarity index 96% rename from veilid-core/src/dht/envelope.rs rename to veilid-core/src/crypto/envelope.rs index d94fbf06..3a3c75e4 100644 --- a/veilid-core/src/dht/envelope.rs +++ b/veilid-core/src/crypto/envelope.rs @@ -1,7 +1,6 @@ #![allow(dead_code)] #![allow(clippy::absurd_extreme_comparisons)] -use super::crypto::*; -use super::key::*; +use super::*; use crate::xx::*; use crate::*; use core::convert::TryInto; @@ -38,8 +37,6 @@ use core::convert::TryInto; pub const MAX_ENVELOPE_SIZE: usize = 65507; pub const MIN_ENVELOPE_SIZE: usize = 0x6A + 0x40; // Header + Signature pub const ENVELOPE_MAGIC: &[u8; 4] = b"VLID"; -pub const MIN_VERSION: u8 = 0u8; -pub const MAX_VERSION: u8 = 0u8; pub type EnvelopeNonce = [u8; 24]; #[derive(Debug, Clone, PartialEq, Eq, Default)] @@ -64,12 +61,12 @@ impl Envelope { assert!(sender_id.valid); assert!(recipient_id.valid); - assert!(version >= MIN_VERSION); - assert!(version <= MAX_VERSION); + assert!(version >= MIN_CRYPTO_VERSION); + assert!(version <= MAX_CRYPTO_VERSION); Self { version, - min_version: MIN_VERSION, - max_version: MAX_VERSION, + min_version: MIN_CRYPTO_VERSION, + max_version: MAX_CRYPTO_VERSION, timestamp, nonce, sender_id, @@ -94,9 +91,9 @@ impl Envelope { // Check version let version = data[0x04]; - if version > MAX_VERSION || version < MIN_VERSION { + if version > MAX_CRYPTO_VERSION || version < MIN_CRYPTO_VERSION { return Err(VeilidAPIError::parse_error( - "unsupported protocol version", + "unsupported cryptography version", version, )); } diff --git a/veilid-core/src/dht/key.rs b/veilid-core/src/crypto/key.rs similarity index 100% rename from veilid-core/src/dht/key.rs rename to veilid-core/src/crypto/key.rs diff --git a/veilid-core/src/dht/crypto.rs b/veilid-core/src/crypto/mod.rs similarity index 97% rename from veilid-core/src/dht/crypto.rs rename to veilid-core/src/crypto/mod.rs index 7680a47b..c7836445 100644 --- a/veilid-core/src/dht/crypto.rs +++ b/veilid-core/src/crypto/mod.rs @@ -1,4 +1,18 @@ -use super::key::*; +mod envelope; +mod key; +mod receipt; +mod value; + +pub mod tests; + +pub use envelope::*; +pub use key::*; +pub use receipt::*; +pub use value::*; + +pub const MIN_CRYPTO_VERSION: u8 = 0u8; +pub const MAX_CRYPTO_VERSION: u8 = 0u8; + use crate::xx::*; use crate::*; use chacha20::cipher::{KeyIvInit, StreamCipher}; diff --git a/veilid-core/src/dht/receipt.rs b/veilid-core/src/crypto/receipt.rs similarity index 97% rename from veilid-core/src/dht/receipt.rs rename to veilid-core/src/crypto/receipt.rs index 74d806cb..b4990b4e 100644 --- a/veilid-core/src/dht/receipt.rs +++ b/veilid-core/src/crypto/receipt.rs @@ -1,7 +1,6 @@ #![allow(dead_code)] #![allow(clippy::absurd_extreme_comparisons)] -use super::envelope::{MAX_VERSION, MIN_VERSION}; -use super::key::*; +use super::*; use crate::xx::*; use crate::*; use core::convert::TryInto; @@ -90,9 +89,9 @@ impl Receipt { // Check version let version = data[0x04]; - if version > MAX_VERSION || version < MIN_VERSION { + if version > MAX_CRYPTO_VERSION || version < MIN_CRYPTO_VERSION { return Err(VeilidAPIError::parse_error( - "unsupported protocol version", + "unsupported cryptography version", version, )); } diff --git a/veilid-core/src/dht/tests/mod.rs b/veilid-core/src/crypto/tests/mod.rs similarity index 100% rename from veilid-core/src/dht/tests/mod.rs rename to veilid-core/src/crypto/tests/mod.rs diff --git a/veilid-core/src/dht/tests/test_crypto.rs b/veilid-core/src/crypto/tests/test_crypto.rs similarity index 100% rename from veilid-core/src/dht/tests/test_crypto.rs rename to veilid-core/src/crypto/tests/test_crypto.rs diff --git a/veilid-core/src/dht/tests/test_dht_key.rs b/veilid-core/src/crypto/tests/test_dht_key.rs similarity index 100% rename from veilid-core/src/dht/tests/test_dht_key.rs rename to veilid-core/src/crypto/tests/test_dht_key.rs diff --git a/veilid-core/src/dht/tests/test_envelope_receipt.rs b/veilid-core/src/crypto/tests/test_envelope_receipt.rs similarity index 100% rename from veilid-core/src/dht/tests/test_envelope_receipt.rs rename to veilid-core/src/crypto/tests/test_envelope_receipt.rs diff --git a/veilid-core/src/dht/value.rs b/veilid-core/src/crypto/value.rs similarity index 100% rename from veilid-core/src/dht/value.rs rename to veilid-core/src/crypto/value.rs diff --git a/veilid-core/src/dht/mod.rs b/veilid-core/src/dht/mod.rs deleted file mode 100644 index 60edad22..00000000 --- a/veilid-core/src/dht/mod.rs +++ /dev/null @@ -1,13 +0,0 @@ -mod crypto; -mod envelope; -mod key; -mod receipt; -mod value; - -pub mod tests; - -pub use crypto::*; -pub use envelope::*; -pub use key::*; -pub use receipt::*; -pub use value::*; diff --git a/veilid-core/src/lib.rs b/veilid-core/src/lib.rs index 222e47b6..1acea791 100644 --- a/veilid-core/src/lib.rs +++ b/veilid-core/src/lib.rs @@ -22,7 +22,7 @@ mod api_tracing_layer; mod attachment_manager; mod callback_state_machine; mod core_context; -mod dht; +mod crypto; mod intf; mod network_manager; mod receipt_manager; diff --git a/veilid-core/src/network_manager/mod.rs b/veilid-core/src/network_manager/mod.rs index 75969de0..a139334b 100644 --- a/veilid-core/src/network_manager/mod.rs +++ b/veilid-core/src/network_manager/mod.rs @@ -22,7 +22,7 @@ pub use network_connection::*; //////////////////////////////////////////////////////////////////////////////////////// use connection_handle::*; use connection_limits::*; -use dht::*; +use crypto::*; use futures_util::stream::{FuturesOrdered, FuturesUnordered, StreamExt}; use hashlink::LruCache; use intf::*; @@ -783,11 +783,7 @@ impl NetworkManager { // Process a received signal #[instrument(level = "trace", skip(self), err)] - pub async fn handle_signal( - &self, - _sender_id: DHTKey, - signal_info: SignalInfo, - ) -> EyreResult> { + pub async fn handle_signal(&self, signal_info: SignalInfo) -> EyreResult> { match signal_info { SignalInfo::ReverseConnect { receipt, peer_info } => { let routing_table = self.routing_table(); @@ -923,7 +919,7 @@ impl NetworkManager { // and if so, get the max version we can use let version = if let Some((node_min, node_max)) = node_ref.min_max_version() { #[allow(clippy::absurd_extreme_comparisons)] - if node_min > MAX_VERSION || node_max < MIN_VERSION { + if node_min > MAX_CRYPTO_VERSION || node_max < MIN_CRYPTO_VERSION { bail!( "can't talk to this node {} because version is unsupported: ({},{})", via_node_id, @@ -931,9 +927,9 @@ impl NetworkManager { node_max ); } - cmp::min(node_max, MAX_VERSION) + cmp::min(node_max, MAX_CRYPTO_VERSION) } else { - MAX_VERSION + MAX_CRYPTO_VERSION }; // Build the envelope to send diff --git a/veilid-core/src/network_manager/tasks.rs b/veilid-core/src/network_manager/tasks.rs index c34b9cb7..5de6568c 100644 --- a/veilid-core/src/network_manager/tasks.rs +++ b/veilid-core/src/network_manager/tasks.rs @@ -1,6 +1,6 @@ use super::*; -use crate::dht::*; +use crate::crypto::*; use crate::xx::*; use futures_util::FutureExt; use stop_token::future::FutureExt as StopFutureExt; @@ -265,8 +265,8 @@ impl NetworkManager { bsmap .entry(node_id) .or_insert_with(|| BootstrapRecord { - min_version: MIN_VERSION, - max_version: MAX_VERSION, + min_version: MIN_CRYPTO_VERSION, + max_version: MAX_CRYPTO_VERSION, dial_info_details: Vec::new(), }) .dial_info_details @@ -299,8 +299,8 @@ impl NetworkManager { network_class: NetworkClass::InboundCapable, // Bootstraps are always inbound capable outbound_protocols: ProtocolTypeSet::only(ProtocolType::UDP), // Bootstraps do not participate in relaying and will not make outbound requests, but will have UDP enabled address_types: AddressTypeSet::all(), // Bootstraps are always IPV4 and IPV6 capable - min_version: v.min_version, // Minimum protocol version specified in txt record - max_version: v.max_version, // Maximum protocol version specified in txt record + min_version: v.min_version, // Minimum crypto version specified in txt record + max_version: v.max_version, // Maximum crypto version specified in txt record dial_info_detail_list: v.dial_info_details, // Dial info is as specified in the bootstrap list relay_peer_info: None, // Bootstraps never require a relay themselves }), diff --git a/veilid-core/src/receipt_manager.rs b/veilid-core/src/receipt_manager.rs index 02bf8a75..06101a3c 100644 --- a/veilid-core/src/receipt_manager.rs +++ b/veilid-core/src/receipt_manager.rs @@ -1,6 +1,6 @@ use crate::*; use core::fmt; -use dht::*; +use crypto::*; use futures_util::stream::{FuturesUnordered, StreamExt}; use network_manager::*; use routing_table::*; diff --git a/veilid-core/src/routing_table/debug.rs b/veilid-core/src/routing_table/debug.rs index 3fa6fbc7..43af8b78 100644 --- a/veilid-core/src/routing_table/debug.rs +++ b/veilid-core/src/routing_table/debug.rs @@ -58,8 +58,8 @@ impl RoutingTable { out += &format!( "{},{},{},{},{}", BOOTSTRAP_TXT_VERSION, - MIN_VERSION, - MAX_VERSION, + MIN_CRYPTO_VERSION, + MAX_CRYPTO_VERSION, self.node_id().encode(), some_hostname.unwrap() ); diff --git a/veilid-core/src/routing_table/mod.rs b/veilid-core/src/routing_table/mod.rs index 3c1ed52c..d11880c5 100644 --- a/veilid-core/src/routing_table/mod.rs +++ b/veilid-core/src/routing_table/mod.rs @@ -9,7 +9,7 @@ mod routing_table_inner; mod stats_accounting; mod tasks; -use crate::dht::*; +use crate::crypto::*; use crate::network_manager::*; use crate::rpc_processor::*; use crate::xx::*; diff --git a/veilid-core/src/routing_table/node_ref.rs b/veilid-core/src/routing_table/node_ref.rs index 1e4385d8..c555c172 100644 --- a/veilid-core/src/routing_table/node_ref.rs +++ b/veilid-core/src/routing_table/node_ref.rs @@ -1,5 +1,5 @@ use super::*; -use crate::dht::*; +use crate::crypto::*; use alloc::fmt; // Connectionless protocols like UDP are dependent on a NAT translation timeout diff --git a/veilid-core/src/routing_table/route_spec_store.rs b/veilid-core/src/routing_table/route_spec_store.rs index 369c3be6..82a1c6a3 100644 --- a/veilid-core/src/routing_table/route_spec_store.rs +++ b/veilid-core/src/routing_table/route_spec_store.rs @@ -49,9 +49,9 @@ struct RouteSpecDetail { /// Directions this route is guaranteed to work in directions: DirectionSet, /// Stability preference (prefer reliable nodes over faster) - stability: Stability, + pub stability: Stability, /// Sequencing preference (connection oriented protocols vs datagram) - sequencing: Sequencing, + pub sequencing: Sequencing, } /// The core representation of the RouteSpecStore that can be serialized @@ -616,11 +616,11 @@ impl RouteSpecStore { routing_table: RoutingTable, safety_selection: SafetySelection, private_route: PrivateRoute, - ) -> Result, RPCError> { + ) -> EyreResult> { let pr_hopcount = private_route.hop_count as usize; let max_route_hop_count = self.max_route_hop_count; if pr_hopcount > max_route_hop_count { - return Err(RPCError::internal("private route hop count too long")); + bail!("private route hop count too long"); } // See if we are using a safety route, if not, short circuit this operation @@ -628,7 +628,7 @@ impl RouteSpecStore { SafetySelection::Unsafe(sequencing) => { // Safety route stub with the node's public key as the safety route key since it's the 0th hop if private_route.first_hop.is_none() { - return Err(RPCError::internal("can't compile zero length route")); + bail!("can't compile zero length route"); } let first_hop = private_route.first_hop.as_ref().unwrap(); let opt_first_hop = match &first_hop.node { @@ -707,10 +707,10 @@ impl RouteSpecStore { // Ensure the total hop count isn't too long for our config let sr_hopcount = safety_spec.hop_count; if sr_hopcount == 0 { - return Err(RPCError::internal("safety route hop count is zero")); + bail!("safety route hop count is zero"); } if sr_hopcount > max_route_hop_count { - return Err(RPCError::internal("safety route hop count too long")); + bail!("safety route hop count too long"); } // See if we can optimize this compilation yet @@ -746,10 +746,10 @@ impl RouteSpecStore { // Encrypt the previous blob ENC(nonce, DH(PKhop,SKsr)) let dh_secret = crypto .cached_dh(&safety_rsd.hops[h], &safety_rsd.secret_key) - .map_err(RPCError::map_internal("dh failed"))?; + .wrap_err("dh failed")?; let enc_msg_data = Crypto::encrypt_aead(blob_data.as_slice(), &nonce, &dh_secret, None) - .map_err(RPCError::map_internal("encryption failed"))?; + .wrap_err("encryption failed")?; // Make route hop data let route_hop_data = RouteHopData { @@ -759,26 +759,23 @@ impl RouteSpecStore { // Make route hop let route_hop = RouteHop { - node: match optimize { + node: if optimize { // Optimized, no peer info, just the dht key - true => RouteNode::NodeId(NodeId::new(safety_rsd.hops[h])), + RouteNode::NodeId(NodeId::new(safety_rsd.hops[h])) + } else { // Full peer info, required until we are sure the route has been fully established - false => { - let node_id = safety_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) - }) + let node_id = safety_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) }) - .flatten(); - if pi.is_none() { - return Err(RPCError::internal( - "peer info should exist for route but doesn't", - )); - } - RouteNode::PeerInfo(pi.unwrap()) + }) + .flatten(); + if pi.is_none() { + bail!("peer info should exist for route but doesn't"); } + RouteNode::PeerInfo(pi.unwrap()) }, next_hop: Some(route_hop_data), }; @@ -838,6 +835,86 @@ impl RouteSpecStore { Ok(Some(compiled_route)) } + /// Assemble private route for publication + pub fn assemble_private_route( + &mut self, + rti: &RoutingTableInner, + routing_table: RoutingTable, + key: &DHTKey, + ) -> EyreResult { + let rsd = self + .detail(&key) + .ok_or_else(|| eyre!("route does not exist"))?; + + // See if we can optimize this compilation yet + // We don't want to include full nodeinfo if we don't have to + let optimize = rsd.reachable; + + // Make innermost route hop to our own node + let mut route_hop = RouteHop { + node: if optimize { + RouteNode::NodeId(NodeId::new(routing_table.node_id())) + } else { + RouteNode::PeerInfo(rti.get_own_peer_info(RoutingDomain::PublicInternet)) + }, + next_hop: None, + }; + + let crypto = routing_table.network_manager().crypto(); + // Loop for each hop + let hop_count = rsd.hops.len(); + for h in (0..hop_count).rev() { + let nonce = Crypto::get_random_nonce(); + + let blob_data = { + let mut rh_message = ::capnp::message::Builder::new_default(); + let mut rh_builder = rh_message.init_root::(); + encode_route_hop(&route_hop, &mut rh_builder)?; + builder_to_vec(rh_message)? + }; + + // Encrypt the previous blob ENC(nonce, DH(PKhop,SKpr)) + let dh_secret = crypto + .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) + .wrap_err("encryption failed")?; + let route_hop_data = RouteHopData { + nonce, + blob: enc_msg_data, + }; + + route_hop = RouteHop { + node: if optimize { + // Optimized, no peer info, just the dht key + RouteNode::NodeId(NodeId::new(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 pi = rti + .with_node_entry(node_id, |entry| { + entry.with(rti, |_rti, e| { + e.make_peer_info(node_id, RoutingDomain::PublicInternet) + }) + }) + .flatten(); + if pi.is_none() { + bail!("peer info should exist for route but doesn't",); + } + RouteNode::PeerInfo(pi.unwrap()) + }, + next_hop: Some(route_hop_data), + } + } + + let private_route = PrivateRoute { + public_key: key.clone(), + hop_count: hop_count.try_into().unwrap(), + first_hop: Some(route_hop), + }; + Ok(private_route) + } + /// Mark route as published /// When first deserialized, routes must be re-published in order to ensure they remain /// in the RouteSpecStore. diff --git a/veilid-core/src/routing_table/routing_domains.rs b/veilid-core/src/routing_table/routing_domains.rs index 7a7cc0e8..b63462a8 100644 --- a/veilid-core/src/routing_table/routing_domains.rs +++ b/veilid-core/src/routing_table/routing_domains.rs @@ -126,8 +126,8 @@ impl RoutingDomainDetailCommon { network_class: self.network_class.unwrap_or(NetworkClass::Invalid), outbound_protocols: self.outbound_protocols, address_types: self.address_types, - min_version: MIN_VERSION, - max_version: MAX_VERSION, + min_version: MIN_CRYPTO_VERSION, + max_version: MAX_CRYPTO_VERSION, dial_info_detail_list: self.dial_info_details.clone(), relay_peer_info: self .relay_node diff --git a/veilid-core/src/rpc_processor/coders/block_id.rs b/veilid-core/src/rpc_processor/coders/block_id.rs index d071197f..1d23f19b 100644 --- a/veilid-core/src/rpc_processor/coders/block_id.rs +++ b/veilid-core/src/rpc_processor/coders/block_id.rs @@ -1,4 +1,4 @@ -use crate::dht::*; +use crate::crypto::*; use crate::*; use core::convert::TryInto; use rpc_processor::*; diff --git a/veilid-core/src/rpc_processor/coders/operations/operation.rs b/veilid-core/src/rpc_processor/coders/operations/operation.rs index fc61cf32..9d36c192 100644 --- a/veilid-core/src/rpc_processor/coders/operations/operation.rs +++ b/veilid-core/src/rpc_processor/coders/operations/operation.rs @@ -19,7 +19,7 @@ impl RPCOperationKind { pub fn decode( kind_reader: &veilid_capnp::operation::kind::Reader, - sender_node_id: &DHTKey, + opt_sender_node_id: Option<&DHTKey>, ) -> Result { let which_reader = kind_reader.which().map_err(RPCError::protocol)?; let out = match which_reader { @@ -30,7 +30,7 @@ impl RPCOperationKind { } veilid_capnp::operation::kind::Which::Statement(r) => { let q_reader = r.map_err(RPCError::protocol)?; - let out = RPCStatement::decode(&q_reader, sender_node_id)?; + let out = RPCStatement::decode(&q_reader, opt_sender_node_id)?; RPCOperationKind::Statement(out) } veilid_capnp::operation::kind::Which::Answer(r) => { @@ -111,22 +111,26 @@ impl RPCOperation { pub fn decode( operation_reader: &veilid_capnp::operation::Reader, - sender_node_id: &DHTKey, + opt_sender_node_id: Option<&DHTKey>, ) -> Result { let op_id = operation_reader.get_op_id(); let sender_node_info = if operation_reader.has_sender_node_info() { - let sni_reader = operation_reader - .get_sender_node_info() - .map_err(RPCError::protocol)?; - let sni = decode_signed_node_info(&sni_reader, sender_node_id, true)?; - Some(sni) + if let Some(sender_node_id) = opt_sender_node_id { + let sni_reader = operation_reader + .get_sender_node_info() + .map_err(RPCError::protocol)?; + let sni = decode_signed_node_info(&sni_reader, sender_node_id, true)?; + Some(sni) + } else { + None + } } else { None }; let kind_reader = operation_reader.get_kind(); - let kind = RPCOperationKind::decode(&kind_reader, sender_node_id)?; + let kind = RPCOperationKind::decode(&kind_reader, opt_sender_node_id)?; Ok(RPCOperation { op_id, diff --git a/veilid-core/src/rpc_processor/coders/operations/operation_node_info_update.rs b/veilid-core/src/rpc_processor/coders/operations/operation_node_info_update.rs index 00c5943a..a785e6b5 100644 --- a/veilid-core/src/rpc_processor/coders/operations/operation_node_info_update.rs +++ b/veilid-core/src/rpc_processor/coders/operations/operation_node_info_update.rs @@ -9,8 +9,14 @@ pub struct RPCOperationNodeInfoUpdate { impl RPCOperationNodeInfoUpdate { pub fn decode( reader: &veilid_capnp::operation_node_info_update::Reader, - sender_node_id: &DHTKey, + opt_sender_node_id: Option<&DHTKey>, ) -> Result { + if opt_sender_node_id.is_none() { + return Err(RPCError::protocol( + "can't decode node info update without sender node id", + )); + } + let sender_node_id = opt_sender_node_id.unwrap(); let sni_reader = reader.get_signed_node_info().map_err(RPCError::protocol)?; let signed_node_info = decode_signed_node_info(&sni_reader, sender_node_id, true)?; diff --git a/veilid-core/src/rpc_processor/coders/operations/operation_route.rs b/veilid-core/src/rpc_processor/coders/operations/operation_route.rs index 7ded216d..68c97191 100644 --- a/veilid-core/src/rpc_processor/coders/operations/operation_route.rs +++ b/veilid-core/src/rpc_processor/coders/operations/operation_route.rs @@ -3,14 +3,16 @@ use rpc_processor::*; #[derive(Debug, Clone)] pub struct RoutedOperation { + pub version: u8, pub signatures: Vec, pub nonce: Nonce, pub data: Vec, } impl RoutedOperation { - pub fn new(nonce: Nonce, data: Vec) -> Self { + pub fn new(version: u8, nonce: Nonce, data: Vec) -> Self { Self { + version, signatures: Vec::new(), nonce, data, @@ -32,11 +34,13 @@ impl RoutedOperation { signatures.push(sig); } + let version = reader.get_version(); let n_reader = reader.get_nonce().map_err(RPCError::protocol)?; let nonce = decode_nonce(&n_reader); let data = reader.get_data().map_err(RPCError::protocol)?.to_vec(); Ok(RoutedOperation { + version, signatures, nonce, data, @@ -47,6 +51,7 @@ impl RoutedOperation { &self, builder: &mut veilid_capnp::routed_operation::Builder, ) -> Result<(), RPCError> { + builder.reborrow().set_version(self.version); let mut sigs_builder = builder.reborrow().init_signatures( self.signatures .len() diff --git a/veilid-core/src/rpc_processor/coders/operations/statement.rs b/veilid-core/src/rpc_processor/coders/operations/statement.rs index bbac6455..96c71a4c 100644 --- a/veilid-core/src/rpc_processor/coders/operations/statement.rs +++ b/veilid-core/src/rpc_processor/coders/operations/statement.rs @@ -22,10 +22,10 @@ impl RPCStatement { } pub fn decode( reader: &veilid_capnp::statement::Reader, - sender_node_id: &DHTKey, + opt_sender_node_id: Option<&DHTKey>, ) -> Result { let d_reader = reader.get_detail(); - let detail = RPCStatementDetail::decode(&d_reader, sender_node_id)?; + let detail = RPCStatementDetail::decode(&d_reader, opt_sender_node_id)?; Ok(RPCStatement { detail }) } pub fn encode(&self, builder: &mut veilid_capnp::statement::Builder) -> Result<(), RPCError> { @@ -59,7 +59,7 @@ impl RPCStatementDetail { } pub fn decode( reader: &veilid_capnp::statement::detail::Reader, - sender_node_id: &DHTKey, + opt_sender_node_id: Option<&DHTKey>, ) -> Result { let which_reader = reader.which().map_err(RPCError::protocol)?; let out = match which_reader { @@ -75,7 +75,7 @@ impl RPCStatementDetail { } veilid_capnp::statement::detail::NodeInfoUpdate(r) => { let op_reader = r.map_err(RPCError::protocol)?; - let out = RPCOperationNodeInfoUpdate::decode(&op_reader, sender_node_id)?; + let out = RPCOperationNodeInfoUpdate::decode(&op_reader, opt_sender_node_id)?; RPCStatementDetail::NodeInfoUpdate(out) } veilid_capnp::statement::detail::ValueChanged(r) => { diff --git a/veilid-core/src/rpc_processor/coders/public_key.rs b/veilid-core/src/rpc_processor/coders/public_key.rs index d80d9401..f7eb530a 100644 --- a/veilid-core/src/rpc_processor/coders/public_key.rs +++ b/veilid-core/src/rpc_processor/coders/public_key.rs @@ -1,4 +1,4 @@ -use crate::dht::*; +use crate::crypto::*; use crate::*; use core::convert::TryInto; use rpc_processor::*; diff --git a/veilid-core/src/rpc_processor/mod.rs b/veilid-core/src/rpc_processor/mod.rs index c426e822..e1742359 100644 --- a/veilid-core/src/rpc_processor/mod.rs +++ b/veilid-core/src/rpc_processor/mod.rs @@ -27,7 +27,7 @@ pub use operation_waiter::*; pub use rpc_error::*; use super::*; -use crate::dht::*; +use crate::crypto::*; use crate::xx::*; use capnp::message::ReaderSegments; use futures_util::StreamExt; @@ -54,7 +54,6 @@ struct RPCMessageHeaderDetailDirect { #[derive(Debug, Clone)] struct RPCMessageHeaderDetailPrivateRoute { - /// The private route we received the rpc over private_route: DHTKey, // The safety selection for replying to this private routed rpc @@ -70,12 +69,6 @@ enum RPCMessageHeaderDetail { /// The decoded header of an RPC message #[derive(Debug, Clone)] struct RPCMessageHeader { - - version: u8, - min_version: u8, - max_version: u8, - timestamp: u64,???? do we need a header for private routed messages? write process_rpc_message - /// Time the message was received, not sent timestamp: u64, /// The length in bytes of the rpc message body @@ -84,6 +77,8 @@ struct RPCMessageHeader { detail: RPCMessageHeaderDetail, } +impl RPCMessageHeader {} + #[derive(Debug)] struct RPCMessageData { contents: Vec, // rpc messages must be a canonicalized single segment @@ -437,6 +432,7 @@ impl RPCProcessor { match self.routing_table().with_route_spec_store_mut(|rss, rti| { // Compile the safety route with the private route rss.compile_safety_route(rti, routing_table, safety_selection, private_route) + .map_err(RPCError::internal) })? { Some(cr) => cr, None => { @@ -448,6 +444,7 @@ impl RPCProcessor { // Encrypt routed operation // Xmsg + ENC(Xmsg, DH(PKapr, SKbsr)) + // xxx use factory method, get version from somewhere... let nonce = Crypto::get_random_nonce(); let dh_secret = self .crypto @@ -457,7 +454,8 @@ impl RPCProcessor { .map_err(RPCError::map_internal("encryption failed"))?; // Make the routed operation - let operation = RoutedOperation::new(nonce, enc_msg_data); + // xxx: replace MAX_CRYPTO_VERSION with the version from the factory + let operation = RoutedOperation::new(MAX_CRYPTO_VERSION, nonce, enc_msg_data); // Prepare route operation let sr_hop_count = compiled_route.safety_route.hop_count; @@ -864,59 +862,79 @@ impl RPCProcessor { ////////////////////////////////////////////////////////////////////// #[instrument(level = "trace", skip(self, encoded_msg), err)] - async fn process_rpc_message_version_0( - &self, - encoded_msg: RPCMessageEncoded, - ) -> Result<(), RPCError> { - // Get the routing domain this message came over - let routing_domain = encoded_msg.header.routing_domain; + async fn process_rpc_message(&self, encoded_msg: RPCMessageEncoded) -> Result<(), RPCError> { + // Decode operation appropriately based on header detail + let msg = match &encoded_msg.header.detail { + RPCMessageHeaderDetail::Direct(detail) => { + // Get the routing domain this message came over + let routing_domain = detail.routing_domain; - // Decode the operation - let sender_node_id = encoded_msg.header.envelope.get_sender_id(); + // Decode the operation + let sender_node_id = detail.envelope.get_sender_id(); - // Decode the RPC message - let operation = { - let reader = capnp::message::Reader::new(encoded_msg.data, Default::default()); - let op_reader = reader - .get_root::() - .map_err(RPCError::protocol) - .map_err(logthru_rpc!())?; - RPCOperation::decode(&op_reader, &sender_node_id)? - }; + // Decode the RPC message + let operation = { + let reader = capnp::message::Reader::new(encoded_msg.data, Default::default()); + let op_reader = reader + .get_root::() + .map_err(RPCError::protocol) + .map_err(logthru_rpc!())?; + RPCOperation::decode(&op_reader, Some(&sender_node_id))? + }; - // Get the sender noderef, incorporating and 'sender node info' - let mut opt_sender_nr: Option = None; - if let Some(sender_node_info) = operation.sender_node_info() { - // Sender NodeInfo was specified, update our routing table with it - if !self.filter_node_info(routing_domain, &sender_node_info.node_info) { - return Err(RPCError::invalid_format( - "sender signednodeinfo has invalid peer scope", - )); + // Get the sender noderef, incorporating and 'sender node info' + let mut opt_sender_nr: Option = None; + if let Some(sender_node_info) = operation.sender_node_info() { + // Sender NodeInfo was specified, update our routing table with it + if !self.filter_node_info(routing_domain, &sender_node_info.node_info) { + return Err(RPCError::invalid_format( + "sender signednodeinfo has invalid peer scope", + )); + } + opt_sender_nr = self.routing_table().register_node_with_signed_node_info( + routing_domain, + sender_node_id, + sender_node_info.clone(), + false, + ); + } + + // look up sender node, in case it's different than our peer due to relaying + if opt_sender_nr.is_none() { + opt_sender_nr = self.routing_table().lookup_node_ref(sender_node_id) + } + + // Mark this sender as having seen our node info over this routing domain + // because it managed to reach us over that routing domain + if let Some(sender_nr) = &opt_sender_nr { + sender_nr.set_seen_our_node_info(routing_domain); + } + + // Make the RPC message + RPCMessage { + header: encoded_msg.header, + operation, + opt_sender_nr, + } } - opt_sender_nr = self.routing_table().register_node_with_signed_node_info( - routing_domain, - sender_node_id, - sender_node_info.clone(), - false, - ); - } + RPCMessageHeaderDetail::PrivateRoute(detail) => { + // Decode the RPC message + let operation = { + let reader = capnp::message::Reader::new(encoded_msg.data, Default::default()); + let op_reader = reader + .get_root::() + .map_err(RPCError::protocol) + .map_err(logthru_rpc!())?; + RPCOperation::decode(&op_reader, None)? + }; - // look up sender node, in case it's different than our peer due to relaying - if opt_sender_nr.is_none() { - opt_sender_nr = self.routing_table().lookup_node_ref(sender_node_id) - } - - // Mark this sender as having seen our node info over this routing domain - // because it managed to reach us over that routing domain - if let Some(sender_nr) = &opt_sender_nr { - sender_nr.set_seen_our_node_info(routing_domain); - } - - // Make the RPC message - let msg = RPCMessage { - header: encoded_msg.header, - operation, - opt_sender_nr, + // Make the RPC message + RPCMessage { + header: encoded_msg.header, + operation, + opt_sender_nr: None, + } + } }; // Process stats @@ -940,7 +958,7 @@ impl RPCProcessor { }; // Log rpc receive - debug!(target: "rpc_message", dir = "recv", kind, op_id = msg.operation.op_id(), desc = msg.operation.kind().desc(), sender_id = ?sender_node_id); + debug!(target: "rpc_message", dir = "recv", kind, op_id = msg.operation.op_id(), desc = msg.operation.kind().desc(), header = ?msg.header); // Process specific message kind match msg.operation.kind() { @@ -977,18 +995,6 @@ impl RPCProcessor { } } - #[instrument(level = "trace", skip(self, msg), err)] - async fn process_rpc_message(&self, msg: RPCMessageEncoded) -> Result<(), RPCError> { - if msg.header.envelope.get_version() == 0 { - self.process_rpc_message_version_0(msg).await - } else { - Err(RPCError::Internal(format!( - "unsupported envelope version: {}, newest supported is version 0", - msg.header.envelope.get_version() - ))) - } - } - async fn rpc_worker( self, stop_token: StopToken, @@ -1044,10 +1050,6 @@ impl RPCProcessor { #[instrument(level = "trace", skip(self, body), err)] pub fn enqueue_private_route_message( &self, - - xx need rpc version somewhere! the rpc version to decode should be packaged in with the body, and the allocator should ensure the version is compatible at the end node. can append to body, and then pop the last u8. - xx try to write the whole process_rpc_message pipeline first - private_route: DHTKey, safety_selection: SafetySelection, body: Vec, diff --git a/veilid-core/src/rpc_processor/rpc_route.rs b/veilid-core/src/rpc_processor/rpc_route.rs index 001bf2bf..91d5aebe 100644 --- a/veilid-core/src/rpc_processor/rpc_route.rs +++ b/veilid-core/src/rpc_processor/rpc_route.rs @@ -72,8 +72,8 @@ impl RPCProcessor { #[instrument(level = "trace", skip_all, err)] async fn process_route_private_route_hop( &self, - route: RPCOperationRoute, - private_route: PrivateRoute, + mut route: RPCOperationRoute, + next_private_route: PrivateRoute, ) -> Result<(), RPCError> { // Make sure hop count makes sense if route.safety_route.hop_count != 0 { @@ -81,19 +81,14 @@ impl RPCProcessor { "Safety hop count should be zero if switched to private route", )); } - if private_route.hop_count as usize > self.unlocked_inner.max_route_hop_count { + if next_private_route.hop_count as usize > self.unlocked_inner.max_route_hop_count { return Err(RPCError::protocol( "Private route hop count too high to process", )); } - if private_route.hop_count == 0 { - return Err(RPCError::protocol( - "Private route hop count should not be zero if there are more hops", - )); - } // Get private route first hop (this is validated to not be None before calling this function) - let first_hop = private_route.first_hop.as_ref().unwrap(); + let first_hop = next_private_route.first_hop.as_ref().unwrap(); // Get next hop node ref let next_hop_nr = match &first_hop.node { @@ -121,12 +116,21 @@ impl RPCProcessor { } }?; + // Sign the operation if this is not our last hop + // as the last hop is already signed by the envelope + if next_private_route.hop_count != 0 { + let node_id = self.routing_table.node_id(); + let node_id_secret = self.routing_table.node_id_secret(); + let sig = sign(&node_id, &node_id_secret, &route.operation.data).map_err(RPCError::internal)?; + route.operation.signatures.push(sig); + } + // Pass along the route let next_hop_route = RPCOperationRoute { safety_route: SafetyRoute { public_key: route.safety_route.public_key, hop_count: 0, - hops: SafetyRouteHops::Private(private_route), + hops: SafetyRouteHops::Private(next_private_route), }, operation: route.operation, }; @@ -146,12 +150,13 @@ impl RPCProcessor { #[instrument(level = "trace", skip_all, err)] async fn process_routed_operation( &self, - sender_id: DHTKey, - route: RPCOperationRoute, + detail: RPCMessageHeaderDetailDirect, + routed_operation: RoutedOperation, + safety_route: &SafetyRoute, private_route: &PrivateRoute, ) -> Result<(), RPCError> { // Make sure hop count makes sense - if route.safety_route.hop_count != 0 { + if safety_route.hop_count != 0 { return Err(RPCError::protocol( "Safety hop count should be zero if switched to private route", )); @@ -162,21 +167,29 @@ impl RPCProcessor { )); } - let routed_operation = &route.operation; - - // Get sequencing preference - if route. - // If the private route public key is our node id, then this was sent via safety route to our node directly // so there will be no signatures to validate let opt_pr_info = if private_route.public_key == self.routing_table.node_id() { // the private route was a stub to our own node's secret - // return our secret key + // return our secret key and an appropriate safety selection + // Get sequencing preference + let sequencing = if detail + .connection_descriptor + .protocol_type() + .is_connection_oriented() + { + Sequencing::EnsureOrdered + } else { + Sequencing::NoPreference + }; Some(( - self.routing_table.node_id_secret(), // Figure out how we'd reply to this if it were a question + self.routing_table.node_id_secret(), SafetySelection::Unsafe(sequencing), )) } else { + // Get sender id + let sender_id = detail.envelope.get_sender_id(); + // Look up the private route and ensure it's one in our spec store let opt_signatures_valid = self.routing_table.with_route_spec_store(|rss, rti| { rss.with_route_spec_detail(&private_route.public_key, |rsd| { @@ -207,10 +220,15 @@ impl RPCProcessor { } } } - // Correct signatures + // We got the correct signatures, return a key ans Some(( rsd.secret_key, - SafetySelection::Safe(SafetySpec { preferred_route: todo!(), hop_count: todo!(), stability: todo!(), sequencing: todo!() }) + SafetySelection::Safe(SafetySpec { + preferred_route: Some(private_route.public_key), + hop_count: rsd.hops.len(), + stability: rsd.stability, + sequencing: rsd.sequencing, + }) )) }) }); @@ -229,7 +247,7 @@ impl RPCProcessor { // xxx: punish nodes that send messages that fail to decrypt eventually let dh_secret = self .crypto - .cached_dh(&route.safety_route.public_key, &secret_key) + .cached_dh(&safety_route.public_key, &secret_key) .map_err(RPCError::protocol)?; let body = Crypto::decrypt_aead( &routed_operation.data, @@ -250,12 +268,14 @@ impl RPCProcessor { #[instrument(level = "trace", skip(self, msg), err)] pub(crate) async fn process_route(&self, msg: RPCMessage) -> Result<(), RPCError> { - // xxx do not process latency for routed messages - // Get header detail, must be direct and not inside a route itself - let (envelope, peer_noderef, connection_descriptor, routing_domain) = match msg.header.detail { - RPCMessageHeaderDetail::Direct { envelope, peer_noderef, connection_descriptor, routing_domain } => (envelope, peer_noderef, connection_descriptor, routing_domain), - RPCMessageHeaderDetail::PrivateRoute { private_route, safety_selection } => { return Err(RPCError::protocol("route operation can not be inside route")) }, + let detail = match msg.header.detail { + RPCMessageHeaderDetail::Direct(detail) => detail, + RPCMessageHeaderDetail::PrivateRoute(_) => { + return Err(RPCError::protocol( + "route operation can not be inside route", + )) + } }; // Get the statement @@ -267,6 +287,14 @@ impl RPCProcessor { _ => panic!("not a statement"), }; + // Process routed operation version + // xxx switch this to a Crypto trait factory method per issue#140 + if route.operation.version != MAX_CRYPTO_VERSION { + return Err(RPCError::protocol( + "routes operation crypto is not valid version", + )); + } + // See what kind of safety route we have going on here match route.safety_route.hops { // There is a safety route hop @@ -312,12 +340,8 @@ impl RPCProcessor { .await?; } else { // Private route is empty, process routed operation - self.process_routed_operation( - envelope.get_sender_id(), - route, - &private_route, - ) - .await?; + self.process_routed_operation(detail, route.operation, &route.safety_route, &private_route) + .await?; } } else if blob_tag == 0 { // RouteHop @@ -373,21 +397,17 @@ impl RPCProcessor { }; // Make next PrivateRoute and pass it on - let private_route = PrivateRoute { + let next_private_route = PrivateRoute { public_key: private_route.public_key, hop_count: private_route.hop_count - 1, first_hop: opt_next_first_hop, }; - self.process_route_private_route_hop(route, private_route) + self.process_route_private_route_hop(route, next_private_route) .await?; } else { // No hops left, time to process the routed operation - self.process_routed_operation( - msg.header.envelope.get_sender_id(), - route, - private_route, - ) - .await?; + self.process_routed_operation(detail, route.operation, &route.safety_route, private_route) + .await?; } } } diff --git a/veilid-core/src/tests/common/mod.rs b/veilid-core/src/tests/common/mod.rs index 537a41a5..ab47cdd5 100644 --- a/veilid-core/src/tests/common/mod.rs +++ b/veilid-core/src/tests/common/mod.rs @@ -7,5 +7,5 @@ pub mod test_veilid_core; use super::*; -pub use dht::tests::*; +pub use crypto::tests::*; pub use network_manager::tests::*; diff --git a/veilid-core/src/tests/native/mod.rs b/veilid-core/src/tests/native/mod.rs index da86b323..d693eaa5 100644 --- a/veilid-core/src/tests/native/mod.rs +++ b/veilid-core/src/tests/native/mod.rs @@ -3,7 +3,7 @@ mod test_async_peek_stream; -use crate::dht::tests::*; +use crate::crypto::tests::*; use crate::network_manager::tests::*; use crate::tests::common::*; use crate::xx::*; diff --git a/veilid-core/src/veilid_api/mod.rs b/veilid-core/src/veilid_api/mod.rs index 9d16b15b..9f14bd68 100644 --- a/veilid-core/src/veilid_api/mod.rs +++ b/veilid-core/src/veilid_api/mod.rs @@ -19,8 +19,8 @@ pub use crate::xx::{ pub use alloc::string::ToString; pub use attachment_manager::AttachmentManager; pub use core::str::FromStr; -pub use dht::Crypto; -pub use dht::{generate_secret, sign, verify, DHTKey, DHTKeySecret, DHTSignature, Nonce}; +pub use crypto::Crypto; +pub use crypto::{generate_secret, sign, verify, DHTKey, DHTKeySecret, DHTSignature, Nonce}; pub use intf::BlockStore; pub use intf::ProtectedStore; pub use intf::TableStore; diff --git a/veilid-core/src/veilid_config.rs b/veilid-core/src/veilid_config.rs index 9fdb88a8..d1287864 100644 --- a/veilid-core/src/veilid_config.rs +++ b/veilid-core/src/veilid_config.rs @@ -708,7 +708,7 @@ impl VeilidConfig { // If we have a node id from storage, check it if node_id.valid && node_id_secret.valid { // Validate node id - if !dht::validate_key(&node_id, &node_id_secret) { + if !crypto::validate_key(&node_id, &node_id_secret) { apibail_generic!("node id secret and node id key don't match"); } }