GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Handle clients with different versions in IPR (#4723)

Browse Source 
* Add signable_request function

* Export key type in function signature

* Cargo.lock

* Track client version and respond using it

* Internally use v7 and then down convert if needed

* Local response type

* Streamline

* Strong type for client version

* Remove commented out code

* rustfmt

* Ignore sign verification fail for v6
This commit is contained in:
Jon Häggblad
2024-07-24 15:35:59 +02:00
committed by GitHub
parent cc25fc1f32
commit 02fde4e530
7 changed files with 357 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/ip-packet-requests/src/v6/conversion.rs
+69
View File
@@ -0,0 +1,69 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::{v6, v7};
impl From<v7::response::StaticConnectFailureReason> for v6::response::StaticConnectFailureReason {
fn from(failure: v7::response::StaticConnectFailureReason) -> Self {
match failure {
v7::response::StaticConnectFailureReason::RequestedIpAlreadyInUse => {
v6::response::StaticConnectFailureReason::RequestedIpAlreadyInUse
}
v7::response::StaticConnectFailureReason::RequestedNymAddressAlreadyInUse => {
v6::response::StaticConnectFailureReason::RequestedNymAddressAlreadyInUse
}
v7::response::StaticConnectFailureReason::OutOfDateTimestamp => {
v6::response::StaticConnectFailureReason::Other("out of date timestamp".to_string())
}
v7::response::StaticConnectFailureReason::Other(reason) => {
v6::response::StaticConnectFailureReason::Other(reason)
}
}
}
}
impl From<v7::response::DynamicConnectFailureReason> for v6::response::DynamicConnectFailureReason {
fn from(failure: v7::response::DynamicConnectFailureReason) -> Self {
match failure {
v7::response::DynamicConnectFailureReason::RequestedNymAddressAlreadyInUse => {
v6::response::DynamicConnectFailureReason::RequestedNymAddressAlreadyInUse
}
v7::response::DynamicConnectFailureReason::NoAvailableIp => {
v6::response::DynamicConnectFailureReason::NoAvailableIp
}
v7::response::DynamicConnectFailureReason::Other(err) => {
v6::response::DynamicConnectFailureReason::Other(err)
}
}
}
}
impl From<v7::response::InfoResponseReply> for v6::response::InfoResponseReply {
fn from(reply: v7::response::InfoResponseReply) -> Self {
match reply {
v7::response::InfoResponseReply::Generic { msg } => {
v6::response::InfoResponseReply::Generic { msg }
}
v7::response::InfoResponseReply::VersionMismatch {
request_version,
response_version,
} => v6::response::InfoResponseReply::VersionMismatch {
request_version,
response_version,
},
v7::response::InfoResponseReply::ExitPolicyFilterCheckFailed { dst } => {
v6::response::InfoResponseReply::ExitPolicyFilterCheckFailed { dst }
}
}
}
}
impl From<v7::response::InfoLevel> for v6::response::InfoLevel {
fn from(level: v7::response::InfoLevel) -> Self {
match level {
v7::response::InfoLevel::Info => v6::response::InfoLevel::Info,
v7::response::InfoLevel::Warn => v6::response::InfoLevel::Warn,
v7::response::InfoLevel::Error => v6::response::InfoLevel::Error,
}
}
}
common/ip-packet-requests/src/v6/mod.rs
+1
View File
@@ -1,3 +1,4 @@
pub mod conversion;
pub mod request;
pub mod response;
common/ip-packet-requests/src/v7/request.rs
+11
View File
@@ -198,6 +198,17 @@ impl IpPacketRequestData {
| IpPacketRequestData::Health(_) => None,
}
}
pub fn signable_request(&self) -> Option<Result<Vec<u8>, SignatureError>> {
match self {
IpPacketRequestData::StaticConnect(request) => Some(request.request()),
IpPacketRequestData::DynamicConnect(request) => Some(request.request()),
IpPacketRequestData::Disconnect(request) => Some(request.request()),
IpPacketRequestData::Data(_) => None,
IpPacketRequestData::Ping(_) => None,
IpPacketRequestData::Health(_) => None,
}
}
}
// A static connect request is when the client provides the internal IP address it will use on the
sdk/rust/nym-sdk/src/mixnet.rs
+1
View File
@@ -67,6 +67,7 @@ pub use nym_credential_storage::{
ephemeral_storage::EphemeralStorage as EphemeralCredentialStorage,
models::StoredIssuedCredential, storage::Storage as CredentialStorage,
};
pub use nym_crypto::asymmetric::ed25519;
pub use nym_network_defaults::NymNetworkDetails;
pub use nym_socks5_client_core::config::Socks5;
pub use nym_sphinx::{
service-providers/ip-packet-router/src/connected_client_handler.rs
+32 -4
View File
@@ -2,12 +2,13 @@
// SPDX-License-Identifier: GPL-3.0-only
use bytes::Bytes;
use nym_ip_packet_requests::{codec::MultiIpPacketCodec, v6::response::IpPacketResponse};
use nym_ip_packet_requests::codec::MultiIpPacketCodec;
use nym_sdk::mixnet::{MixnetMessageSender, Recipient};
use crate::{
constants::CLIENT_HANDLER_ACTIVITY_TIMEOUT,
error::{IpPacketRouterError, Result},
mixnet_listener::SupportedClientVersion,
util::create_message::create_input_message,
};
@@ -18,13 +19,29 @@ use crate::{
// This handler is spawned as a task, and it listens to IP packets passed from the tun_listener,
// encodes it, and then sends to mixnet.
pub(crate) struct ConnectedClientHandler {
// The address of the client that this handler is connected to
nym_address: Recipient,
// The number of hops the packet should take before reaching the client
mix_hops: Option<u8>,
// Channel to receive packets from the tun_listener
forward_from_tun_rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
// Channel to send packets to the mixnet
mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender,
// Channel to receive close signal
close_rx: tokio::sync::oneshot::Receiver<()>,
// Interval to check for activity timeout
activity_timeout: tokio::time::Interval,
// Encoder to bundle multiple packets into a single one
encoder: MultiIpPacketCodec,
// The version of the client
client_version: SupportedClientVersion,
}
impl ConnectedClientHandler {
@@ -32,6 +49,7 @@ impl ConnectedClientHandler {
reply_to: Recipient,
reply_to_hops: Option<u8>,
buffer_timeout: std::time::Duration,
client_version: SupportedClientVersion,
mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender,
) -> (
tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
@@ -55,6 +73,7 @@ impl ConnectedClientHandler {
close_rx,
activity_timeout,
encoder,
client_version,
};
let handle = tokio::spawn(async move {
@@ -67,9 +86,18 @@ impl ConnectedClientHandler {
}
async fn send_packets_to_mixnet(&mut self, packets: Bytes) -> Result<()> {
let response_packet = IpPacketResponse::new_ip_packet(packets)
.to_bytes()
.map_err(|err| IpPacketRouterError::FailedToSerializeResponsePacket { source: err })?;
let response_packet = match self.client_version {
SupportedClientVersion::V6 => {
nym_ip_packet_requests::v6::response::IpPacketResponse::new_ip_packet(packets)
.to_bytes()
}
SupportedClientVersion::V7 => {
nym_ip_packet_requests::v7::response::IpPacketResponse::new_ip_packet(packets)
.to_bytes()
}
}
.map_err(|err| IpPacketRouterError::FailedToSerializeResponsePacket { source: err })?;
let input_message = create_input_message(self.nym_address, response_packet, self.mix_hops);
self.mixnet_client_sender
service-providers/ip-packet-router/src/error.rs
+3
View File
@@ -96,6 +96,9 @@ pub enum IpPacketRouterError {
FailedToVerifyRequest {
source: nym_ip_packet_requests::v7::signature::SignatureError,
},
#[error("client is connected with an invalid version: {version}")]
InvalidConnectedClientVersion { version: u8 },
}
pub type Result<T> = std::result::Result<T, IpPacketRouterError>;
service-providers/ip-packet-router/src/mixnet_listener.rs
+240 -67
View File
@@ -4,22 +4,19 @@ use std::{collections::HashMap, net::SocketAddr};
use bytes::{Bytes, BytesMut};
use futures::StreamExt;
use nym_ip_packet_requests::v7::response::{
DynamicConnectFailureReason, InfoLevel, InfoResponseReply, StaticConnectFailureReason,
};
use nym_ip_packet_requests::{
codec::MultiIpPacketCodec,
v6::{
self,
response::{
DynamicConnectFailureReason, InfoLevel, InfoResponseReply, IpPacketResponse,
StaticConnectFailureReason,
},
},
v6,
v7::{
self,
request::{
DataRequest, DisconnectRequest, DynamicConnectRequest, IpPacketRequest,
IpPacketRequestData, StaticConnectRequest,
},
signature::{SignatureError, SignedRequest},
signature::SignedRequest,
},
IpPair,
};
@@ -266,7 +263,132 @@ impl Drop for CloseTx {
}
}
type PacketHandleResult = Result<Option<v6::response::IpPacketResponse>>;
type PacketHandleResult = Result<Option<Response>>;
#[derive(Debug, Clone)]
enum Response {
V6(v6::response::IpPacketResponse),
V7(v7::response::IpPacketResponse),
}
impl Response {
fn recipient(&self) -> Option<&Recipient> {
match self {
Response::V6(response) => response.recipient(),
Response::V7(response) => response.recipient(),
}
}
fn new_static_connect_success(
request_id: u64,
reply_to: Recipient,
client_version: SupportedClientVersion,
) -> Self {
match client_version {
SupportedClientVersion::V6 => Response::V6(
v6::response::IpPacketResponse::new_static_connect_success(request_id, reply_to),
),
SupportedClientVersion::V7 => Response::V7(
v7::response::IpPacketResponse::new_static_connect_success(request_id, reply_to),
),
}
}
fn new_static_connect_failure(
request_id: u64,
reply_to: Recipient,
reason: StaticConnectFailureReason,
client_version: SupportedClientVersion,
) -> Self {
match client_version {
SupportedClientVersion::V6 => {
Response::V6(v6::response::IpPacketResponse::new_static_connect_failure(
request_id,
reply_to,
reason.into(),
))
}
SupportedClientVersion::V7 => {
Response::V7(v7::response::IpPacketResponse::new_static_connect_failure(
request_id, reply_to, reason,
))
}
}
}
fn new_dynamic_connect_success(
request_id: u64,
reply_to: Recipient,
ips: IpPair,
client_version: SupportedClientVersion,
) -> Self {
match client_version {
SupportedClientVersion::V6 => {
Response::V6(v6::response::IpPacketResponse::new_dynamic_connect_success(
request_id, reply_to, ips,
))
}
SupportedClientVersion::V7 => {
Response::V7(v7::response::IpPacketResponse::new_dynamic_connect_success(
request_id, reply_to, ips,
))
}
}
}
fn new_dynamic_connect_failure(
request_id: u64,
reply_to: Recipient,
reason: DynamicConnectFailureReason,
client_version: SupportedClientVersion,
) -> Self {
match client_version {
SupportedClientVersion::V6 => {
Response::V6(v6::response::IpPacketResponse::new_dynamic_connect_failure(
request_id,
reply_to,
reason.into(),
))
}
SupportedClientVersion::V7 => {
Response::V7(v7::response::IpPacketResponse::new_dynamic_connect_failure(
request_id, reply_to, reason,
))
}
}
}
fn new_data_info_response(
reply_to: Recipient,
reply: InfoResponseReply,
level: InfoLevel,
client_version: SupportedClientVersion,
) -> Self {
match client_version {
SupportedClientVersion::V6 => {
Response::V6(v6::response::IpPacketResponse::new_data_info_response(
reply_to,
reply.into(),
level.into(),
))
}
SupportedClientVersion::V7 => Response::V7(
v7::response::IpPacketResponse::new_data_info_response(reply_to, reply, level),
),
}
}
fn to_bytes(&self) -> Result<Vec<u8>> {
match self {
Response::V6(response) => response.to_bytes(),
Response::V7(response) => response.to_bytes(),
}
.map_err(|err| {
log::error!("Failed to serialize response packet");
IpPacketRouterError::FailedToSerializeResponsePacket { source: err }
})
}
}
#[cfg(target_os = "linux")]
pub(crate) struct MixnetListener {
@@ -297,6 +419,7 @@ impl MixnetListener {
async fn on_static_connect_request(
&mut self,
connect_request: StaticConnectRequest,
client_version: SupportedClientVersion,
) -> PacketHandleResult {
log::info!(
"Received static connect request from {sender_address}",
@@ -328,8 +451,10 @@ impl MixnetListener {
{
log::error!("Failed to update activity for client");
};
Ok(Some(IpPacketResponse::new_static_connect_success(
request_id, reply_to,
Ok(Some(Response::new_static_connect_success(
request_id,
reply_to,
client_version,
)))
}
(false, false) => {
@@ -341,6 +466,7 @@ impl MixnetListener {
reply_to,
reply_to_hops,
buffer_timeout,
client_version,
self.mixnet_client.split_sender(),
);
@@ -353,24 +479,28 @@ impl MixnetListener {
close_tx,
handle,
);
Ok(Some(IpPacketResponse::new_static_connect_success(
request_id, reply_to,
Ok(Some(Response::new_static_connect_success(
request_id,
reply_to,
client_version,
)))
}
(true, false) => {
log::info!("Requested IP is not available");
Ok(Some(IpPacketResponse::new_static_connect_failure(
Ok(Some(Response::new_static_connect_failure(
request_id,
reply_to,
StaticConnectFailureReason::RequestedIpAlreadyInUse,
client_version,
)))
}
(false, true) => {
log::info!("Nym address is already registered");
Ok(Some(IpPacketResponse::new_static_connect_failure(
Ok(Some(Response::new_static_connect_failure(
request_id,
reply_to,
StaticConnectFailureReason::RequestedNymAddressAlreadyInUse,
client_version,
)))
}
}
@@ -379,6 +509,7 @@ impl MixnetListener {
async fn on_dynamic_connect_request(
&mut self,
connect_request: DynamicConnectRequest,
client_version: SupportedClientVersion,
) -> PacketHandleResult {
log::info!(
"Received dynamic connect request from {sender_address}",
@@ -406,19 +537,21 @@ impl MixnetListener {
{
log::error!("Failed to update activity for client");
}
return Ok(Some(IpPacketResponse::new_dynamic_connect_success(
return Ok(Some(Response::new_dynamic_connect_success(
request_id,
reply_to,
existing_ips,
client_version,
)));
}
let Some(new_ips) = self.connected_clients.find_new_ip() else {
log::info!("No available IP address");
return Ok(Some(IpPacketResponse::new_dynamic_connect_failure(
return Ok(Some(Response::new_dynamic_connect_failure(
request_id,
reply_to,
DynamicConnectFailureReason::NoAvailableIp,
client_version,
)));
};
@@ -428,6 +561,7 @@ impl MixnetListener {
reply_to,
reply_to_hops,
buffer_timeout,
client_version,
self.mixnet_client.split_sender(),
);
@@ -440,17 +574,28 @@ impl MixnetListener {
close_tx,
handle,
);
Ok(Some(IpPacketResponse::new_dynamic_connect_success(
request_id, reply_to, new_ips,
Ok(Some(Response::new_dynamic_connect_success(
request_id,
reply_to,
new_ips,
client_version,
)))
}
fn on_disconnect_request(&self, _disconnect_request: DisconnectRequest) -> PacketHandleResult {
fn on_disconnect_request(
&self,
_disconnect_request: DisconnectRequest,
_client_version: SupportedClientVersion,
) -> PacketHandleResult {
log::info!("Received disconnect request: not implemented, dropping");
Ok(None)
}
async fn handle_packet(&mut self, ip_packet: &Bytes) -> PacketHandleResult {
async fn handle_packet(
&mut self,
ip_packet: &Bytes,
client_version: SupportedClientVersion,
) -> PacketHandleResult {
log::trace!("Received data request");
// We don't forward packets that we are not able to parse. BUT, there might be a good
@@ -487,12 +632,13 @@ impl MixnetListener {
Ok(None)
} else {
log::info!("Denied filter check: {dst}");
Ok(Some(IpPacketResponse::new_data_info_response(
Ok(Some(Response::new_data_info_response(
connected_client.nym_address,
InfoResponseReply::ExitPolicyFilterCheckFailed {
dst: dst.to_string(),
},
InfoLevel::Warn,
client_version,
)))
}
} else {
@@ -505,13 +651,14 @@ impl MixnetListener {
async fn on_data_request(
&mut self,
data_request: DataRequest,
client_version: SupportedClientVersion,
) -> Result<Vec<PacketHandleResult>> {
let mut responses = Vec::new();
let mut decoder = MultiIpPacketCodec::new(nym_ip_packet_requests::codec::BUFFER_TIMEOUT);
let mut bytes = BytesMut::new();
bytes.extend_from_slice(&data_request.ip_packets);
while let Ok(Some(packet)) = decoder.decode(&mut bytes) {
let result = self.handle_packet(&packet).await;
let result = self.handle_packet(&packet, client_version).await;
responses.push(result);
}
Ok(responses)
@@ -519,26 +666,13 @@ impl MixnetListener {
fn on_version_mismatch(
&self,
version: u8,
reconstructed: &ReconstructedMessage,
_version: u8,
_reconstructed: &ReconstructedMessage,
) -> PacketHandleResult {
// If it's possible to parse, do so and return back a response, otherwise just drop
let (id, recipient) =
v6::request::IpPacketRequest::from_reconstructed_message(reconstructed)
.ok()
.and_then(|request| {
request
.recipient()
.map(|recipient| (request.id().unwrap_or(0), *recipient))
})
.ok_or(IpPacketRouterError::InvalidPacketVersion(version))?;
Ok(Some(IpPacketResponse::new_version_mismatch(
id,
recipient,
version,
nym_ip_packet_requests::CURRENT_VERSION,
)))
// Just drop it. In the future we might want to return a response here, if for example
// the client is connecting with a version that is older than the currently supported
// ones.
Ok(None)
}
async fn on_reconstructed_message(
@@ -550,7 +684,7 @@ impl MixnetListener {
reconstructed.sender_tag
);
let request = match deserialize_request(&reconstructed) {
let (request, client_version) = match deserialize_request(&reconstructed) {
Err(IpPacketRouterError::InvalidPacketVersion(version)) => {
return Ok(vec![self.on_version_mismatch(version, &reconstructed)]);
}
@@ -559,21 +693,31 @@ impl MixnetListener {
match request.data {
IpPacketRequestData::StaticConnect(signed_connect_request) => {
verify_signed_request(&signed_connect_request)?;
verify_signed_request(&signed_connect_request, client_version)?;
let connect_request = signed_connect_request.request;
Ok(vec![self.on_static_connect_request(connect_request).await])
Ok(vec![
self.on_static_connect_request(connect_request, client_version)
.await,
])
}
IpPacketRequestData::DynamicConnect(signed_connect_request) => {
verify_signed_request(&signed_connect_request)?;
verify_signed_request(&signed_connect_request, client_version)?;
let connect_request = signed_connect_request.request;
Ok(vec![self.on_dynamic_connect_request(connect_request).await])
Ok(vec![
self.on_dynamic_connect_request(connect_request, client_version)
.await,
])
}
IpPacketRequestData::Disconnect(signed_disconnect_request) => {
verify_signed_request(&signed_disconnect_request)?;
verify_signed_request(&signed_disconnect_request, client_version)?;
let disconnect_request = signed_disconnect_request.request;
Ok(vec![self.on_disconnect_request(disconnect_request)])
Ok(vec![
self.on_disconnect_request(disconnect_request, client_version)
])
}
IpPacketRequestData::Data(data_request) => {
self.on_data_request(data_request, client_version).await
}
IpPacketRequestData::Data(data_request) => self.on_data_request(data_request).await,
IpPacketRequestData::Ping(_) => {
log::info!("Received ping request: not implemented, dropping");
Ok(vec![])
@@ -605,16 +749,13 @@ impl MixnetListener {
// When an incoming mixnet message triggers a response that we send back, such as during
// connect handshake.
async fn handle_response(&self, response: IpPacketResponse) -> Result<()> {
async fn handle_response(&self, response: Response) -> Result<()> {
let Some(recipient) = response.recipient() else {
log::error!("No recipient in response packet, this should NOT happen!");
return Err(IpPacketRouterError::NoRecipientInResponse);
};
let response_packet = response.to_bytes().map_err(|err| {
log::error!("Failed to serialize response packet");
IpPacketRouterError::FailedToSerializeResponsePacket { source: err }
})?;
let response_packet = response.to_bytes()?;
// We could avoid this lookup if we check this when we create the response.
let mix_hops = if let Some(c) = self
@@ -687,33 +828,65 @@ impl MixnetListener {
}
}
fn deserialize_request(reconstructed: &ReconstructedMessage) -> Result<IpPacketRequest> {
fn deserialize_request(
reconstructed: &ReconstructedMessage,
) -> Result<(IpPacketRequest, SupportedClientVersion)> {
let request_version = *reconstructed
.message
.first()
.ok_or(IpPacketRouterError::EmptyPacket)?;
// Check version of the request and convert to the latest version if necessary
match request_version {
6 => v6::request::IpPacketRequest::from_reconstructed_message(reconstructed)
.map_err(|err| IpPacketRouterError::FailedToDeserializeTaggedPacket { source: err })
.map(|r| r.into()),
7 => v7::request::IpPacketRequest::from_reconstructed_message(reconstructed)
.map_err(|err| IpPacketRouterError::FailedToDeserializeTaggedPacket { source: err }),
let request = match request_version {
6 => nym_ip_packet_requests::v6::request::IpPacketRequest::from_reconstructed_message(
reconstructed,
)
.map_err(|err| IpPacketRouterError::FailedToDeserializeTaggedPacket { source: err })
.map(|r| r.into()),
7 => nym_ip_packet_requests::v7::request::IpPacketRequest::from_reconstructed_message(
reconstructed,
)
.map_err(|err| IpPacketRouterError::FailedToDeserializeTaggedPacket { source: err }),
_ => {
log::info!("Received packet with invalid version: v{request_version}");
Err(IpPacketRouterError::InvalidPacketVersion(request_version))
}
};
let Some(request_version) = SupportedClientVersion::new(request_version) else {
return Err(IpPacketRouterError::InvalidPacketVersion(request_version));
};
// Tag the request with the version of the request
request.map(|r| (r, request_version))
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub(crate) enum SupportedClientVersion {
V6,
V7,
}
impl SupportedClientVersion {
fn new(request_version: u8) -> Option<Self> {
match request_version {
6 => Some(SupportedClientVersion::V6),
7 => Some(SupportedClientVersion::V7),
_ => None,
}
}
}
fn verify_signed_request(request: &impl SignedRequest) -> Result<()> {
fn verify_signed_request(
request: &impl SignedRequest,
client_version: SupportedClientVersion,
) -> Result<()> {
if let Err(err) = request.verify() {
// Once we start to require clients to send v7 requests, we will enfore checking
// signatures. Until then, we only check if they are present.
if !matches!(err, SignatureError::MissingSignature) {
return Err(IpPacketRouterError::FailedToVerifyRequest { source: err });
// If the client is V6, we don't care about missing signature
if client_version == SupportedClientVersion::V6 {
return Ok(());
}
return Err(IpPacketRouterError::FailedToVerifyRequest { source: err });
}
Ok(())
}