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

checkpoint

Browse Source
This commit is contained in:
John Smith
2022-10-30 19:29:31 -04:00
parent d94a023c32
commit 50718b7074
35 changed files with 334 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
veilid-core/src/rpc_processor/coders/block_id.rs
View File

@@ -1,4 +1,4 @@
use crate::dht::*;
use crate::crypto::*;
use crate::*;
use core::convert::TryInto;
use rpc_processor::*;

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

@@ -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<Self, RPCError> {
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<Self, RPCError> {
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,

8
veilid-core/src/rpc_processor/coders/operations/operation_node_info_update.rs
View File

@@ -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<RPCOperationNodeInfoUpdate, RPCError> {
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)?;

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

@@ -3,14 +3,16 @@ use rpc_processor::*;
#[derive(Debug, Clone)]
pub struct RoutedOperation {
pub version: u8,
pub signatures: Vec<DHTSignature>,
pub nonce: Nonce,
pub data: Vec<u8>,
}
impl RoutedOperation {
pub fn new(nonce: Nonce, data: Vec<u8>) -> Self {
pub fn new(version: u8, nonce: Nonce, data: Vec<u8>) -> 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()

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

@@ -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<RPCStatement, RPCError> {
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<RPCStatementDetail, RPCError> {
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) => {

2
veilid-core/src/rpc_processor/coders/public_key.rs
View File

@@ -1,4 +1,4 @@
use crate::dht::*;
use crate::crypto::*;
use crate::*;
use core::convert::TryInto;
use rpc_processor::*;

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

@@ -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<u8>, // 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::<veilid_capnp::operation::Reader>()
.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::<veilid_capnp::operation::Reader>()
.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<NodeRef> = 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<NodeRef> = 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::<veilid_capnp::operation::Reader>()
.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<u8>,

104
veilid-core/src/rpc_processor/rpc_route.rs
View File

@@ -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?;
}
}
}