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

checkpoint

Browse Source
This commit is contained in:
John Smith
2023-02-17 17:47:21 -05:00
parent ca3ce91365
commit 05be3099c8
17 changed files with 205 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,12 +16,15 @@ impl RPCOperationKind {
}
}
pub fn decode(kind_reader: &veilid_capnp::operation::kind::Reader) -> Result<Self, RPCError> {
pub fn decode(
kind_reader: &veilid_capnp::operation::kind::Reader,
crypto: Crypto,
) -> Result<Self, RPCError> {
let which_reader = kind_reader.which().map_err(RPCError::protocol)?;
let out = match which_reader {
veilid_capnp::operation::kind::Which::Question(r) => {
let q_reader = r.map_err(RPCError::protocol)?;
let out = RPCQuestion::decode(&q_reader)?;
let out = RPCQuestion::decode(&q_reader, crypto)?;
RPCOperationKind::Question(out)
}
veilid_capnp::operation::kind::Which::Statement(r) => {
@@ -54,31 +57,25 @@ impl RPCOperationKind {
#[derive(Debug, Clone)]
pub struct RPCOperation {
op_id: OperationId,
sender_node_info: Option<SignedNodeInfo>,
opt_sender_peer_info: Option<PeerInfo>,
target_node_info_ts: Timestamp,
kind: RPCOperationKind,
}
impl RPCOperation {
pub fn new_question(
question: RPCQuestion,
sender_signed_node_info: SenderSignedNodeInfo,
) -> Self {
pub fn new_question(question: RPCQuestion, sender_peer_info: SenderPeerInfo) -> Self {
Self {
op_id: OperationId::new(get_random_u64()),
sender_node_info: sender_signed_node_info.signed_node_info,
target_node_info_ts: sender_signed_node_info.target_node_info_ts,
opt_sender_peer_info: sender_peer_info.opt_sender_peer_info,
target_node_info_ts: sender_peer_info.target_node_info_ts,
kind: RPCOperationKind::Question(question),
}
}
pub fn new_statement(
statement: RPCStatement,
sender_signed_node_info: SenderSignedNodeInfo,
) -> Self {
pub fn new_statement(statement: RPCStatement, sender_peer_info: SenderPeerInfo) -> Self {
Self {
op_id: OperationId::new(get_random_u64()),
sender_node_info: sender_signed_node_info.signed_node_info,
target_node_info_ts: sender_signed_node_info.target_node_info_ts,
opt_sender_peer_info: sender_peer_info.opt_sender_peer_info,
target_node_info_ts: sender_peer_info.target_node_info_ts,
kind: RPCOperationKind::Statement(statement),
}
}
@@ -86,12 +83,12 @@ impl RPCOperation {
pub fn new_answer(
request: &RPCOperation,
answer: RPCAnswer,
sender_signed_node_info: SenderSignedNodeInfo,
sender_peer_info: SenderPeerInfo,
) -> Self {
Self {
op_id: request.op_id,
sender_node_info: sender_signed_node_info.signed_node_info,
target_node_info_ts: sender_signed_node_info.target_node_info_ts,
opt_sender_peer_info: sender_peer_info.opt_sender_peer_info,
target_node_info_ts: sender_peer_info.target_node_info_ts,
kind: RPCOperationKind::Answer(answer),
}
}
@@ -100,8 +97,8 @@ impl RPCOperation {
self.op_id
}
pub fn sender_node_info(&self) -> Option<&SignedNodeInfo> {
self.sender_node_info.as_ref()
pub fn sender_peer_info(&self) -> Option<&PeerInfo> {
self.opt_sender_peer_info.as_ref()
}
pub fn target_node_info_ts(&self) -> Timestamp {
self.target_node_info_ts
@@ -117,20 +114,16 @@ impl RPCOperation {
pub fn decode(
operation_reader: &veilid_capnp::operation::Reader,
opt_sender_node_id: Option<&TypedKey>,
crypto: Crypto,
) -> Result<Self, RPCError> {
let op_id = OperationId::new(operation_reader.get_op_id());
let sender_node_info = if operation_reader.has_sender_node_info() {
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)?;
Some(sni)
} else {
None
}
let sender_peer_info = if operation_reader.has_sender_peer_info() {
let pi_reader = operation_reader
.get_sender_peer_info()
.map_err(RPCError::protocol)?;
let pi = decode_peer_info(&pi_reader, crypto.clone())?;
Some(pi)
} else {
None
};
@@ -138,11 +131,11 @@ impl RPCOperation {
let target_node_info_ts = Timestamp::new(operation_reader.get_target_node_info_ts());
let kind_reader = operation_reader.get_kind();
let kind = RPCOperationKind::decode(&kind_reader)?;
let kind = RPCOperationKind::decode(&kind_reader, crypto)?;
Ok(RPCOperation {
op_id,
sender_node_info,
opt_sender_peer_info: sender_peer_info,
target_node_info_ts,
kind,
})
@@ -150,9 +143,9 @@ impl RPCOperation {
pub fn encode(&self, builder: &mut veilid_capnp::operation::Builder) -> Result<(), RPCError> {
builder.set_op_id(self.op_id.as_u64());
if let Some(sender_info) = &self.sender_node_info {
let mut si_builder = builder.reborrow().init_sender_node_info();
encode_signed_node_info(&sender_info, &mut si_builder)?;
if let Some(sender_peer_info) = &self.opt_sender_peer_info {
let mut pi_builder = builder.reborrow().init_sender_peer_info();
encode_peer_info(&sender_peer_info, &mut pi_builder)?;
}
builder.set_target_node_info_ts(self.target_node_info_ts.as_u64());
let mut k_builder = builder.reborrow().init_kind();

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

@@ -19,9 +19,12 @@ impl RPCQuestion {
pub fn desc(&self) -> &'static str {
self.detail.desc()
}
pub fn decode(reader: &veilid_capnp::question::Reader) -> Result<RPCQuestion, RPCError> {
pub fn decode(
reader: &veilid_capnp::question::Reader,
crypto: Crypto,
) -> Result<RPCQuestion, RPCError> {
let rt_reader = reader.get_respond_to();
let respond_to = RespondTo::decode(&rt_reader)?;
let respond_to = RespondTo::decode(&rt_reader, crypto)?;
let d_reader = reader.get_detail();
let detail = RPCQuestionDetail::decode(&d_reader)?;
Ok(RPCQuestion { respond_to, detail })

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

@@ -23,12 +23,15 @@ impl RespondTo {
Ok(())
}
pub fn decode(reader: &veilid_capnp::question::respond_to::Reader) -> Result<Self, RPCError> {
pub fn decode(
reader: &veilid_capnp::question::respond_to::Reader,
crypto: Crypto,
) -> Result<Self, RPCError> {
let respond_to = match reader.which().map_err(RPCError::protocol)? {
veilid_capnp::question::respond_to::Sender(()) => RespondTo::Sender,
veilid_capnp::question::respond_to::PrivateRoute(pr_reader) => {
let pr_reader = pr_reader.map_err(RPCError::protocol)?;
let pr = decode_private_route(&pr_reader)?;
let pr = decode_private_route(&pr_reader, crypto)?;
RespondTo::PrivateRoute(pr)
}
};

35
veilid-core/src/rpc_processor/coders/private_safety_route.rs
View File

@@ -53,7 +53,7 @@ pub fn encode_route_hop(
match &route_hop.node {
RouteNode::NodeId(ni) => {
let mut ni_builder = node_builder.init_node_id();
encode_key256(&ni.key, &mut ni_builder)?;
encode_key256(&ni, &mut ni_builder);
}
RouteNode::PeerInfo(pi) => {
let mut pi_builder = node_builder.init_peer_info();
@@ -67,17 +67,20 @@ pub fn encode_route_hop(
Ok(())
}
pub fn decode_route_hop(reader: &veilid_capnp::route_hop::Reader) -> Result<RouteHop, RPCError> {
pub fn decode_route_hop(
reader: &veilid_capnp::route_hop::Reader,
crypto: Crypto,
) -> Result<RouteHop, RPCError> {
let n_reader = reader.reborrow().get_node();
let node = match n_reader.which().map_err(RPCError::protocol)? {
veilid_capnp::route_hop::node::Which::NodeId(ni) => {
let ni_reader = ni.map_err(RPCError::protocol)?;
RouteNode::NodeId(NodeId::new(decode_key256(&ni_reader)))
RouteNode::NodeId(decode_key256(&ni_reader))
}
veilid_capnp::route_hop::node::Which::PeerInfo(pi) => {
let pi_reader = pi.map_err(RPCError::protocol)?;
RouteNode::PeerInfo(
decode_peer_info(&pi_reader)
decode_peer_info(&pi_reader, crypto)
.map_err(RPCError::map_protocol("invalid peer info in route hop"))?,
)
}
@@ -101,10 +104,10 @@ pub fn encode_private_route(
private_route: &PrivateRoute,
builder: &mut veilid_capnp::private_route::Builder,
) -> Result<(), RPCError> {
encode_key256(
encode_typed_key(
&private_route.public_key,
&mut builder.reborrow().init_public_key(),
)?;
);
builder.set_hop_count(private_route.hop_count);
let mut h_builder = builder.reborrow().init_hops();
match &private_route.hops {
@@ -125,16 +128,17 @@ pub fn encode_private_route(
pub fn decode_private_route(
reader: &veilid_capnp::private_route::Reader,
crypto: Crypto,
) -> Result<PrivateRoute, RPCError> {
let public_key = decode_key256(&reader.get_public_key().map_err(RPCError::map_protocol(
"invalid public key in private route",
))?);
let public_key = decode_typed_key(&reader.get_public_key().map_err(
RPCError::map_protocol("invalid public key in private route"),
)?)?;
let hop_count = reader.get_hop_count();
let hops = match reader.get_hops().which().map_err(RPCError::protocol)? {
veilid_capnp::private_route::hops::Which::FirstHop(rh_reader) => {
let rh_reader = rh_reader.map_err(RPCError::protocol)?;
PrivateRouteHops::FirstHop(decode_route_hop(&rh_reader)?)
PrivateRouteHops::FirstHop(decode_route_hop(&rh_reader, crypto)?)
}
veilid_capnp::private_route::hops::Which::Data(rhd_reader) => {
let rhd_reader = rhd_reader.map_err(RPCError::protocol)?;
@@ -156,10 +160,10 @@ pub fn encode_safety_route(
safety_route: &SafetyRoute,
builder: &mut veilid_capnp::safety_route::Builder,
) -> Result<(), RPCError> {
encode_key256(
encode_typed_key(
&safety_route.public_key,
&mut builder.reborrow().init_public_key(),
)?;
);
builder.set_hop_count(safety_route.hop_count);
let h_builder = builder.reborrow().init_hops();
match &safety_route.hops {
@@ -178,12 +182,13 @@ pub fn encode_safety_route(
pub fn decode_safety_route(
reader: &veilid_capnp::safety_route::Reader,
crypto: Crypto,
) -> Result<SafetyRoute, RPCError> {
let public_key = decode_key256(
let public_key = decode_typed_key(
&reader
.get_public_key()
.map_err(RPCError::map_protocol("invalid public key in safety route"))?,
);
)?;
let hop_count = reader.get_hop_count();
let hops = match reader.get_hops().which().map_err(RPCError::protocol)? {
veilid_capnp::safety_route::hops::Which::Data(rhd_reader) => {
@@ -192,7 +197,7 @@ pub fn decode_safety_route(
}
veilid_capnp::safety_route::hops::Which::Private(pr_reader) => {
let pr_reader = pr_reader.map_err(RPCError::protocol)?;
SafetyRouteHops::Private(decode_private_route(&pr_reader)?)
SafetyRouteHops::Private(decode_private_route(&pr_reader, crypto)?)
}
};

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

@@ -162,8 +162,9 @@ impl RPCProcessor {
}
SafetySelection::Safe(safety_spec) => {
// 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(safety_spec, &target.node_ids())
.get_private_route_for_safety_spec(ck, safety_spec, &target.node_ids().keys())
.map_err(RPCError::internal)? else {
return Ok(NetworkResult::no_connection_other("no private route for response at this time"));
};
@@ -187,10 +188,12 @@ impl RPCProcessor {
}
SafetySelection::Safe(safety_spec) => {
// Sent via a relay but with a safety route, respond to private route
let ck = target.best_node_id().kind;
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(safety_spec, &avoid_nodes)
.get_private_route_for_safety_spec(ck, safety_spec, &avoid_nodes.keys())
.map_err(RPCError::internal)? else {
return Ok(NetworkResult::no_connection_other("no private route for response at this time"));
};
@@ -219,13 +222,13 @@ impl RPCProcessor {
// Determine if we can use optimized nodeinfo
let route_node = match rss
.has_remote_private_route_seen_our_node_info(&private_route.public_key)
.has_remote_private_route_seen_our_node_info(&private_route.public_key.key)
{
true => {
if !routing_table.has_valid_own_node_info(RoutingDomain::PublicInternet) {
return Ok(NetworkResult::no_connection_other("Own node info must be valid to use private route"));
}
RouteNode::NodeId(routing_table.node_id(crypto_kind))
RouteNode::NodeId(routing_table.node_id(crypto_kind).key)
}
false => {
let Some(own_peer_info) =
@@ -245,14 +248,14 @@ impl RPCProcessor {
// Check for loopback test
let pr_key = if safety_spec.preferred_route
== Some(private_route.public_key)
== Some(private_route.public_key.key)
{
// Private route is also safety route during loopback test
private_route.public_key
private_route.public_key.key
} else {
// Get the privat route to respond to that matches the safety route spec we sent the request with
// Get the private route to respond to that matches the safety route spec we sent the request with
let Some(pr_key) = rss
.get_private_route_for_safety_spec(safety_spec, &[avoid_node_id])
.get_private_route_for_safety_spec(crypto_kind, safety_spec, &[avoid_node_id])
.map_err(RPCError::internal)? else {
return Ok(NetworkResult::no_connection_other("no private route for response at this time"));
};
@@ -300,18 +303,25 @@ impl RPCProcessor {
};
// Reply directly to the request's source
let sender_id = TypedKey::new(detail.envelope.get_crypto_kind(), detail.envelope.get_sender_id();
let sender_node_id = TypedKey::new(detail.envelope.get_crypto_kind(), detail.envelope.get_sender_id());
// This may be a different node's reference than the 'sender' in the case of a relay
let peer_noderef = detail.peer_noderef.clone();
// If the sender_id is that of the peer, then this is a direct reply
// else it is a relayed reply through the peer
xxx continue here, make sure respond to semantics are correct
if peer_noderef.node_id() == sender_id {
if peer_noderef.node_ids().contains(&sender_node_id) {
NetworkResult::value(Destination::direct(peer_noderef))
} else {
NetworkResult::value(Destination::relay(peer_noderef, sender_id))
// Look up the sender node, we should have added it via senderNodeInfo before getting here.
if let Some(sender_noderef) = self.routing_table.lookup_node_ref(sender_node_id) {
NetworkResult::value(Destination::relay(peer_noderef, sender_noderef))
} else {
return NetworkResult::invalid_message(
"not responding to sender that has no node info",
);
}
}
}
RespondTo::PrivateRoute(pr) => {

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

@@ -179,22 +179,22 @@ struct RenderedOperation {
/// Node information exchanged during every RPC message
#[derive(Default, Debug, Clone)]
pub struct SenderSignedNodeInfo {
/// The current signed node info of the sender if required
signed_node_info: Option<SignedNodeInfo>,
pub struct SenderPeerInfo {
/// The current peer info of the sender if required
opt_sender_peer_info: Option<PeerInfo>,
/// The last timestamp of the target's node info to assist remote node with sending its latest node info
target_node_info_ts: Timestamp,
}
impl SenderSignedNodeInfo {
pub fn new_no_sni(target_node_info_ts: Timestamp) -> Self {
impl SenderPeerInfo {
pub fn new_no_peer_info(target_node_info_ts: Timestamp) -> Self {
Self {
signed_node_info: None,
opt_sender_peer_info: None,
target_node_info_ts,
}
}
pub fn new(sender_signed_node_info: SignedNodeInfo, target_node_info_ts: Timestamp) -> Self {
pub fn new(sender_peer_info: PeerInfo, target_node_info_ts: Timestamp) -> Self {
Self {
signed_node_info: Some(sender_signed_node_info),
opt_sender_peer_info: Some(sender_peer_info),
target_node_info_ts,
}
}
@@ -686,17 +686,14 @@ impl RPCProcessor {
/// Get signed node info to package with RPC messages to improve
/// routing table caching when it is okay to do so
#[instrument(level = "trace", skip(self), ret, err)]
fn get_sender_signed_node_info(
&self,
dest: &Destination,
) -> Result<SenderSignedNodeInfo, RPCError> {
fn get_sender_signed_node_info(&self, dest: &Destination) -> Result<SenderPeerInfo, RPCError> {
// Don't do this if the sender is to remain private
// Otherwise we would be attaching the original sender's identity to the final destination,
// thus defeating the purpose of the safety route entirely :P
match dest.get_safety_selection() {
SafetySelection::Unsafe(_) => {}
SafetySelection::Safe(_) => {
return Ok(SenderSignedNodeInfo::default());
return Ok(SenderPeerInfo::default());
}
}
@@ -716,14 +713,14 @@ impl RPCProcessor {
target
} else {
// Target was not in our routing table
return Ok(SenderSignedNodeInfo::default());
return Ok(SenderPeerInfo::default());
}
}
Destination::PrivateRoute {
private_route: _,
safety_selection: _,
} => {
return Ok(SenderSignedNodeInfo::default());
return Ok(SenderPeerInfo::default());
}
};
@@ -737,7 +734,7 @@ impl RPCProcessor {
// Don't return our node info if it's not valid yet
let Some(own_peer_info) = routing_table.get_own_peer_info(routing_domain) else {
return Ok(SenderSignedNodeInfo::new_no_sni(target_node_info_ts));
return Ok(SenderPeerInfo::new_no_peer_info(target_node_info_ts));
};
// Get our node info timestamp
@@ -745,10 +742,10 @@ impl RPCProcessor {
// If the target has seen our node info already don't send it again
if target.has_seen_our_node_info_ts(routing_domain, our_node_info_ts) {
return Ok(SenderSignedNodeInfo::new_no_sni(target_node_info_ts));
return Ok(SenderPeerInfo::new_no_peer_info(target_node_info_ts));
}
Ok(SenderSignedNodeInfo::new(
Ok(SenderPeerInfo::new(
own_peer_info.signed_node_info,
target_node_info_ts,
))
@@ -1205,19 +1202,20 @@ impl RPCProcessor {
RPCOperation::decode(&op_reader, Some(&sender_node_id))?
};
// Get the sender noderef, incorporating and 'sender node info'
// Get the sender noderef, incorporating sender's peer 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) {
if let Some(sender_peer_info) = operation.sender_peer_info() {
// Ensure the sender peer info is for the actual sender specified in the envelope
// Sender PeerInfo was specified, update our routing table with it
if !self.filter_node_info(routing_domain, &sender_peer_info) {
return Err(RPCError::invalid_format(
"sender signednodeinfo has invalid peer scope",
"sender peerinfo has invalid peer scope",
));
}
opt_sender_nr = self.routing_table().register_node_with_signed_node_info(
opt_sender_nr = self.routing_table().register_node_with_peer_info(
routing_domain,
sender_node_id,
sender_node_info.clone(),
sender_peer_info,
false,
);
}