Merge pull request #4 from expressvpn/cvpn-555

CVPN-555: Add support for plugins to send a reply packet

lightway-client/src/io/inside/tun.rs

@@ -9,7 +9,7 @@ use pnet::packet::ipv4::Ipv4Packet;
 use lightway_app_utils::Tun as AppUtilsTun;
 use lightway_core::{
     ipv4_update_destination, ipv4_update_source, IOCallbackResult, InsideIOSendCallback,
-    InsideIOSendCallbackArg, SessionId,
+    InsideIOSendCallbackArg, InsideIpConfig,
 };
 
 use crate::{io::inside::InsideIO, ConnectionState};
@@ -34,6 +34,26 @@ impl Tun {
         };
         Ok(Tun { tun, ip, dns_ip })
     }
+
+    /// Api to send packet in the tunnel
+    pub fn try_send(&self, mut pkt: BytesMut, ip_config: Option<InsideIpConfig>) -> Result<usize> {
+        let pkt_len = pkt.len();
+        // Update destination IP from server provided inside ip to TUN device ip
+        ipv4_update_destination(pkt.as_mut(), self.ip);
+
+        // Update source IP from server DNS ip to TUN DNS ip
+        if let Some(ip_config) = ip_config {
+            let packet = Ipv4Packet::new(pkt.as_ref());
+            if let Some(packet) = packet {
+                if packet.get_source() == ip_config.dns_ip {
+                    ipv4_update_source(pkt.as_mut(), self.dns_ip);
+                }
+            };
+        }
+
+        self.tun.try_send(pkt);
+        Ok(pkt_len)
+    }
 }
 
 #[async_trait]
@@ -48,12 +68,7 @@ impl InsideIO for Tun {
 }
 
 impl InsideIOSendCallback<ConnectionState> for Tun {
-    fn send(
-        &self,
-        _session_id: SessionId,
-        mut buf: BytesMut,
-        state: &mut ConnectionState,
-    ) -> IOCallbackResult<usize> {
+    fn send(&self, mut buf: BytesMut, state: &mut ConnectionState) -> IOCallbackResult<usize> {
         // Update destination IP from server provided inside ip to TUN device ip
         ipv4_update_destination(buf.as_mut(), self.ip);
 

lightway-client/src/lib.rs

@@ -329,7 +329,8 @@ pub async fn client(config: ClientConfig) -> Result<()> {
             let mut conn = conn.lock().unwrap();
 
             // Update source IP address to server assigned IP address
-            if let Some(ip_config) = conn.app_state().ip_config {
+            let ip_config = conn.app_state().ip_config;
+            if let Some(ip_config) = &ip_config {
                 ipv4_update_source(buf.as_mut(), ip_config.client_ip);
 
                 // Update TUN device DNS IP address to server provided DNS address
@@ -343,6 +344,10 @@ pub async fn client(config: ClientConfig) -> Result<()> {
 
             match conn.inside_data_received(buf) {
                 Ok(()) => {}
+                Err(ConnectionError::PluginDropWithReply(reply)) => {
+                    // Send the reply packet to inside path
+                    let _ = inside_io.try_send(reply, ip_config);
+                }
                 Err(ConnectionError::InvalidState) => {
                     // Ignore the packet till the connection is online
                 }

lightway-core/src/connection.rs

@@ -145,6 +145,10 @@ pub enum ConnectionError {
     #[error("Invalid inside packet: {0}")]
     InvalidInsidePacket(InvalidPacketError),
 
+    /// Plugin returns a reply packet
+    #[error("Plugin dropped with a reply packet")]
+    PluginDropWithReply(BytesMut),
+
     /// Plugin returns error
     #[error("Plugin error: {0}")]
     PluginError(Box<dyn std::error::Error + Sync + Send>),
@@ -193,6 +197,7 @@ impl ConnectionError {
                     UnknownSessionID => false,
                     InvalidInsidePacket(_) => false,
                     RejectedSessionID => false,
+                    PluginDropWithReply(_) => false,
                     PluginError(_) => false,
                     PacketError(_) => false,
                     DataFragmentError(_) => false,
@@ -726,6 +731,9 @@ impl<AppState: Send> Connection<AppState> {
             PluginResult::Drop => {
                 return Ok(());
             }
+            PluginResult::DropWithReply(b) => {
+                return Err(ConnectionError::PluginDropWithReply(b));
+            }
             PluginResult::Error(e) => {
                 return Err(ConnectionError::PluginError(e));
             }
@@ -1291,6 +1299,9 @@ impl<AppState: Send> Connection<AppState> {
             PluginResult::Drop => {
                 return Ok(());
             }
+            PluginResult::DropWithReply(b) => {
+                return Err(ConnectionError::PluginDropWithReply(b));
+            }
             PluginResult::Error(e) => {
                 return Err(ConnectionError::PluginError(e));
             }
@@ -1301,7 +1312,7 @@ impl<AppState: Send> Connection<AppState> {
         match self
             .inside_io
             .clone()
-            .send(self.session_id, inside_pkt, self.app_state_mut())
+            .send(inside_pkt, self.app_state_mut())
         {
             IOCallbackResult::Ok(_nr) => {}
             IOCallbackResult::Err(err) => {

lightway-core/src/connection/io_adapter.rs

@@ -154,6 +154,10 @@ impl WolfSSLIOAdapter {
             PluginResult::Drop => {
                 return IOCallbackResult::Ok(buf.len());
             }
+            // Outside plugins cannot drop with reply
+            PluginResult::DropWithReply(_) => {
+                return IOCallbackResult::Ok(buf.len());
+            }
             PluginResult::Error(e) => {
                 use std::io::{Error, ErrorKind};
                 return IOCallbackResult::Err(Error::new(ErrorKind::Other, e));
@@ -221,6 +225,11 @@ impl WolfSSLIOAdapter {
                     send_buffer.complete();
                     return IOCallbackResult::Ok(buf.len());
                 }
+                // Outside plugins cannot drop with reply
+                PluginResult::DropWithReply(_) => {
+                    send_buffer.complete();
+                    return IOCallbackResult::Ok(buf.len());
+                }
                 PluginResult::Error(e) => {
                     use std::io::{Error, ErrorKind};
                     send_buffer.complete();

lightway-core/src/io.rs

@@ -1,6 +1,5 @@
 use std::{net::SocketAddr, sync::Arc};
 
-use crate::SessionId;
 use bytes::BytesMut;
 use wolfssl::IOCallbackResult;
 
@@ -12,12 +11,7 @@ pub trait InsideIOSendCallback<AppState> {
     /// return the number of bytes actually consumed. If the operation would
     /// block [`std::io::ErrorKind::WouldBlock`] then return
     /// [`IOCallbackResult::WouldBlock`].
-    fn send(
-        &self,
-        session_id: SessionId,
-        buf: BytesMut,
-        state: &mut AppState,
-    ) -> IOCallbackResult<usize>;
+    fn send(&self, buf: BytesMut, state: &mut AppState) -> IOCallbackResult<usize>;
 
     /// MTU supported by this inside I/O path
     fn mtu(&self) -> usize;

lightway-core/src/packet.rs

@@ -40,6 +40,7 @@ impl OutsidePacket {
                 let mut buf = match plugins.do_ingress(&mut buf) {
                     PluginResult::Accept => buf,
                     PluginResult::Drop => return Err(OutsidePacketError::PluginDrop),
+                    PluginResult::DropWithReply(_) => return Err(OutsidePacketError::PluginDrop),
                     PluginResult::Error(e) => return Err(OutsidePacketError::PluginError(e)),
                 };
 

lightway-core/src/plugin.rs

@@ -42,6 +42,11 @@ pub enum PluginResult {
     /// [`Plugin`] dropped the packet
     #[error("Plugin drops the packet")]
     Drop,
+    /// [`Plugin`] dropped the packet and returned a reply packet to send back
+    /// This is useful only for Inside IO plugins. Outside plugins cannot
+    /// drop a packet with reply
+    #[error("Plugin drops the packet with reply packet")]
+    DropWithReply(BytesMut),
     /// Internal [`Plugin`] error
     #[error("Plugin error")]
     Error(Box<dyn std::error::Error + Sync + Send>),

lightway-core/tests/connection.rs

@@ -40,12 +40,7 @@ impl ChannelTun {
     }
 }
 impl InsideIOSendCallback<ConnectionTicker> for ChannelTun {
-    fn send(
-        &self,
-        _session_id: SessionId,
-        buf: BytesMut,
-        _state: &mut ConnectionTicker,
-    ) -> IOCallbackResult<usize> {
+    fn send(&self, buf: BytesMut, _state: &mut ConnectionTicker) -> IOCallbackResult<usize> {
         let buf_len = buf.len();
         self.0.send(buf.freeze()).expect("Send");
         IOCallbackResult::Ok(buf_len)

lightway-server/src/io/inside/tun.rs

@@ -6,11 +6,10 @@ use async_trait::async_trait;
 use bytes::BytesMut;
 use lightway_app_utils::Tun as AppUtilsTun;
 use lightway_core::{
-    ipv4_update_source, IOCallbackResult, InsideIOSendCallback, InsideIOSendCallbackArg, SessionId,
+    ipv4_update_source, IOCallbackResult, InsideIOSendCallback, InsideIOSendCallbackArg,
 };
 use std::os::fd::{AsRawFd, RawFd};
 use std::sync::Arc;
-use tracing::warn;
 
 pub(crate) struct Tun(AppUtilsTun);
 
@@ -48,14 +47,9 @@ impl InsideIO for Tun {
 }
 
 impl InsideIOSendCallback<ConnectionState> for Tun {
-    fn send(
-        &self,
-        session_id: SessionId,
-        mut buf: BytesMut,
-        state: &mut ConnectionState,
-    ) -> IOCallbackResult<usize> {
+    fn send(&self, mut buf: BytesMut, state: &mut ConnectionState) -> IOCallbackResult<usize> {
         let Some(client_ip) = state.ip else {
-            warn!(?session_id, "Unable to find client ip");
+            metrics::tun_rejected_packet_no_client_ip();
             // Ip address not found, dropping the packet
             return IOCallbackResult::Ok(buf.len());
         };

lightway-server/src/metrics.rs

@@ -35,6 +35,7 @@ const METRIC_TUN_REJECTED_INVALID_STATE: &str = "tun_rejected_packet_invalid_sta
 const METRIC_TUN_REJECTED_INVALID_INSIDE_PACKET: &str = "tun_rejected_packet_invalid_inside_packet";
 const METRIC_TUN_REJECTED_OTHER: &str = "tun_rejected_packet_invalid_other";
 const METRIC_TUN_REJECTED_NO_CONNECTION: &str = "tun_rejected_packet_no_connection";
+const METRIC_TUN_REJECTED_NO_CLIENT_IP: &str = "tun_rejected_packet_no_client_ip";
 
 // Traffic volume
 const METRIC_TUN_FROM_CLIENT: &str = "tun_from_client";
@@ -211,6 +212,11 @@ pub(crate) fn tun_rejected_packet_no_connection() {
     counter!(METRIC_TUN_REJECTED_NO_CONNECTION).increment(1);
 }
 
+/// Tunnel rejected packet, since there was no clientip in app state
+pub(crate) fn tun_rejected_packet_no_client_ip() {
+    counter!(METRIC_TUN_REJECTED_NO_CLIENT_IP).increment(1);
+}
+
 /// Bytes sent from client to the TUN device.
 pub(crate) fn tun_from_client(sz: usize) {
     counter!(METRIC_TUN_FROM_CLIENT).increment(sz as u64);