work

src/main.rs

@@ -4,14 +4,18 @@ mod cake;
 mod config;
 mod netlink;
 mod pinger;
+mod pinger_icmp;
+mod pinger_icmp_ts;
 
 use std::net::IpAddr;
 use std::str::FromStr;
-use std::{thread, time};
+use std::{process, thread, time};
 
 use crate::config::Config;
-use crate::netlink::{find_interface, find_qdisc, set_qdisc_rate};
-use crate::pinger::{Pinger, PingerICMPEcho};
+use crate::netlink::{find_interface, find_qdisc, get_interface_stats, set_qdisc_rate};
+use crate::pinger::Pinger;
+use crate::pinger_icmp::PingerICMPEcho;
+use crate::pinger_icmp_ts::PingerICMPTimestamps;
 
 #[macro_export]
 macro_rules! error {
@@ -30,7 +34,7 @@ fn main() {
     println!("starting up");
 
     let config = Config::new();
-    test(config);
+    test(&config);
 
     let mut reflectors: Vec<IpAddr> = Vec::new();
 
@@ -39,18 +43,30 @@ fn main() {
     reflectors.push(IpAddr::from_str("9.9.9.9").unwrap());
     reflectors.push(IpAddr::from_str("9.9.9.10").unwrap());
 
-    let pinger = PingerICMPEcho::new(reflectors);
-    let pinger_receiver = pinger.clone();
-    let pinger_sender = pinger.clone();
+    get_interface_stats(config.upload_interface.as_str()).unwrap();
 
-    let receiver_handle = thread::spawn(move || pinger_receiver.receive_loop());
-    let sender_handle = thread::spawn(move || pinger_sender.sender_loop());
+    let mut pinger_receiver =
+        PingerICMPTimestamps::new((process::id() & 0xFFFF) as u16, reflectors.clone());
+    let mut pinger_sender =
+        PingerICMPTimestamps::new((process::id() & 0xFFFF) as u16, reflectors.clone());
 
-    receiver_handle.join().unwrap();
-    sender_handle.join().unwrap();
+    let receiver_handle = thread::Builder::new()
+        .name("receiver".to_string())
+        .spawn(move || pinger_receiver.receive_loop())
+        .unwrap();
+    let sender_handle = thread::Builder::new()
+        .name("sender".to_string())
+        .spawn(move || pinger_sender.sender_loop())
+        .unwrap();
+
+    let threads = vec![receiver_handle, sender_handle];
+
+    for thread in threads {
+        thread.join().unwrap();
+    }
 }
 
-fn test(config: Config) {
+fn test(config: &Config) {
     if true {
         return;
     }

src/netlink.rs

@@ -8,9 +8,11 @@ use neli::nl::{NlPayload, Nlmsghdr};
 use neli::rtnl::{Ifinfomsg, Rtattr, Tcmsg};
 use neli::socket::NlSocketHandle;
 use neli::types::{Buffer, RtBuffer};
+use serde::Deserialize;
 use std::error::Error;
 use std::fmt::Display;
 
+use bincode::deserialize;
 use std::fmt;
 
 #[derive(Default, Debug)]
@@ -30,6 +32,34 @@ impl Display for NoQdiscFoundError {
     }
 }
 
+#[derive(Deserialize, Copy, Clone, Default, Debug)]
+pub struct RtnlLinkStats64 {
+    pub rx_packets: u64,
+    pub tx_packets: u64,
+    pub rx_bytes: u64,
+    pub tx_bytes: u64,
+    pub rx_errors: u64,
+    pub tx_errors: u64,
+    pub rx_dropped: u64,
+    pub tx_dropped: u64,
+    pub multicast: u64,
+    pub collisions: u64,
+    pub rx_length_errors: u64,
+    pub rx_over_errors: u64,
+    pub rx_crc_errors: u64,
+    pub rx_frame_errors: u64,
+    pub rx_fifo_errors: u64,
+    pub rx_missed_errors: u64,
+    pub tx_aborted_errors: u64,
+    pub tx_carrier_errors: u64,
+    pub tx_fifo_errors: u64,
+    pub tx_heartbeat_errors: u64,
+    pub tx_window_errors: u64,
+    pub rx_compressed: u64,
+    pub tx_compressed: u64,
+    pub rx_nohandler: u64,
+}
+
 #[allow(dead_code)]
 pub struct Qdisc {
     ifindex: i32,
@@ -39,8 +69,7 @@ pub struct Qdisc {
     parent: u32,
 }
 
-pub fn find_interface(ifname: &str) -> Result<i32, Box<dyn Error>> {
-    let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap();
+fn nl_interface_get(socket: &mut NlSocketHandle, ifname: &str) -> Result<(), Box<dyn Error>> {
     let mut attrs = RtBuffer::new();
 
     const RTEXT_FILTER_VF: i32 = 1 << 0;
@@ -70,6 +99,14 @@ pub fn find_interface(ifname: &str) -> Result<i32, Box<dyn Error>> {
 
     socket.send(nlhdr)?;
 
+    Ok(())
+}
+
+pub fn find_interface(ifname: &str) -> Result<i32, Box<dyn Error>> {
+    let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap();
+
+    nl_interface_get(&mut socket, ifname).expect("Wrong when sending intf");
+
     for response in socket.iter(false) {
         let header: Nlmsghdr<Rtm, Ifinfomsg> = response?;
 
@@ -86,6 +123,40 @@ pub fn find_interface(ifname: &str) -> Result<i32, Box<dyn Error>> {
     Ok(-1)
 }
 
+pub fn get_interface_stats(ifname: &str) -> Result<RtnlLinkStats64, Box<dyn Error>> {
+    let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap();
+
+    nl_interface_get(&mut socket, ifname).expect("Wrong when sending intf");
+
+    for response in socket.iter(false) {
+        let header: Nlmsghdr<Rtm, Ifinfomsg> = response?;
+
+        if header.nl_type != Rtm::Newlink {
+            return Err(Box::new(NlError::msg("Netlink error retrieving link")));
+        }
+
+        if let NlPayload::Payload(p) = header.nl_payload {
+            for attr in p.rtattrs.iter() {
+                //println!("{:?}", attr.rta_type);
+
+                if attr.rta_type == Ifla::Stats64 {
+                    println!("{:?}", attr.rta_payload.as_ref());
+                    let buf = attr.rta_payload.as_ref();
+
+                    let stats: RtnlLinkStats64 = deserialize(buf).unwrap();
+
+                    println!("stats rx: {}", stats.rx_bytes);
+                    println!("stats tx: {}", stats.tx_bytes);
+
+                    return Ok(stats);
+                }
+            }
+        }
+    }
+
+    Err(Box::new(NlError::msg("Error getting stats")))
+}
+
 pub fn find_qdisc(ifindex: i32) -> Result<Qdisc, Box<dyn Error>> {
     let mut socket = NlSocketHandle::connect(NlFamily::Route, None, &[]).unwrap();
     let tc_msg = Tcmsg::new(

src/pinger.rs

@@ -1,215 +1,9 @@
-use byteorder::{BigEndian, ReadBytesExt};
-use nix::sys::socket::{
-    recvfrom, sendto, AddressFamily, MsgFlags, SockFlag, SockProtocol, SockType, SockaddrIn,
-    SockaddrIn6, SockaddrLike,
-};
-use nix::sys::time::TimeValLike;
-use nix::time::{clock_gettime, ClockId};
 use std::error::Error;
-use std::mem::size_of;
-use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
-use std::os::unix::io::RawFd;
-use std::str::FromStr;
-use std::{thread, time};
-
-struct ICMP {
-    icmp_type: u8,
-    code: u8,
-    checksum: u16,
-    identifier: u16,
-    sequence: u16,
-    payload: u64,
-}
-
-impl ICMP {
-    fn to_bytes(&self) -> Box<[u8; 16]> {
-        let checksum = self.checksum.to_be_bytes();
-        let identifier = self.identifier.to_be_bytes();
-        let sequence = self.sequence.to_be_bytes();
-        let payload = self.payload.to_be_bytes();
-
-        let buf = [
-            self.icmp_type,
-            self.code,
-            checksum[0],
-            checksum[1],
-            identifier[0],
-            identifier[1],
-            sequence[0],
-            sequence[1],
-            payload[0],
-            payload[1],
-            payload[2],
-            payload[3],
-            payload[4],
-            payload[5],
-            payload[6],
-            payload[7],
-        ];
-
-        return Box::new(buf);
-    }
-}
-
-impl From<&[u8]> for ICMP {
-    fn from(buf: &[u8]) -> Self {
-        ICMP {
-            icmp_type: buf[0],
-            code: buf[1],
-            checksum: (&buf[2..3]).read_u16::<BigEndian>().unwrap(),
-            identifier: (&buf[4..5]).read_u16::<BigEndian>().unwrap(),
-            sequence: (&buf[6..7]).read_u16::<BigEndian>().unwrap(),
-            payload: (&buf[8..16]).read_u64::<BigEndian>().unwrap(),
-        }
-    }
-}
-
-impl From<&mut [u8]> for ICMP {
-    fn from(buf: &mut [u8]) -> Self {
-        ICMP {
-            icmp_type: buf[0],
-            code: buf[1],
-            checksum: (&buf[2..3]).read_u16::<BigEndian>().unwrap(),
-            identifier: (&buf[4..5]).read_u16::<BigEndian>().unwrap(),
-            sequence: (&buf[6..7]).read_u16::<BigEndian>().unwrap(),
-            payload: (&buf[8..16]).read_u64::<BigEndian>().unwrap(),
-        }
-    }
-}
+use std::net::IpAddr;
 
 pub trait Pinger {
-    fn new(reflectors: Vec<IpAddr>) -> Self;
-    fn receive_loop(&self);
-    fn receive_ping(&self) -> Result<(), Box<dyn Error>>;
-    fn sender_loop(&self);
-    fn send_ping(&self, reflector: &IpAddr, seq: u16) -> Result<(), Box<dyn Error>>;
-}
-
-#[derive(Clone)]
-pub struct PingerICMPEcho {
-    reflectors: Vec<IpAddr>,
-    socket: RawFd,
-}
-
-impl PingerICMPEcho {
-    fn calculate_checksum(buffer: &mut [u8]) -> u16 {
-        let mut sum = 0u32;
-        for word in buffer.chunks(2) {
-            let mut part = u16::from(word[0]) << 8;
-            if word.len() > 1 {
-                part += u16::from(word[1]);
-            }
-            sum = sum.wrapping_add(u32::from(part));
-        }
-
-        while (sum >> 16) > 0 {
-            sum = (sum & 0xffff) + (sum >> 16);
-        }
-
-        !sum as u16
-    }
-}
-
-impl Pinger for PingerICMPEcho {
-    fn new(reflectors: Vec<IpAddr>) -> Self {
-        let socket = nix::sys::socket::socket(
-            AddressFamily::Inet,
-            SockType::Raw,
-            SockFlag::empty(),
-            SockProtocol::ICMP,
-        )
-        .expect("Couldn't open a socket");
-
-        PingerICMPEcho { reflectors, socket }
-    }
-
-    fn receive_loop(&self) {
-        loop {
-            let mut v: Vec<u8> = vec![0; size_of::<ICMP>()];
-            let buf = v.as_mut_slice();
-
-            println!("buf len: {}", buf.len());
-
-            let res = recvfrom::<SockaddrIn>(self.socket, buf).unwrap();
-            let remote = res.1.unwrap();
-
-            let ip = IpAddr::from(remote.ip().to_ne_bytes());
-
-            if self.reflectors.contains(&ip) {
-                println!("ip: {}", ip.to_string());
-                println!("usize: {}", res.0);
-                println!("{:?}", buf);
-
-                let hdr: ICMP = buf.into();
-                println!("Type: {}", hdr.icmp_type);
-            }
-        }
-    }
-
-    fn receive_ping(&self) -> Result<(), Box<dyn Error>> {
-        // Process ping RTT
-        /*let ping_rtt = SystemTime::now()
-        .duration_since(ping_start_time)
-        .unwrap_or(Duration::from_secs(0));*/
-        todo!()
-    }
-
-    fn sender_loop(&self) {
-        let mut seq: u16 = 0;
-        let tick_duration_ms: u16 = 500;
-        let sleep_duration = time::Duration::from_millis(tick_duration_ms as u64);
-
-        loop {
-            for reflector in self.reflectors.iter() {
-                self.send_ping(reflector, seq).expect("TODO: panic message");
-            }
-
-            thread::sleep(sleep_duration);
-
-            seq += 1;
-
-            if seq > std::u16::MAX {
-                seq = 0;
-            }
-        }
-    }
-
-    fn send_ping(&self, reflector: &IpAddr, seq: u16) -> Result<(), Box<dyn Error>> {
-        let time = clock_gettime(ClockId::CLOCK_MONOTONIC).unwrap();
-
-        let mut hdr = ICMP {
-            icmp_type: 8,
-            code: 0,
-            checksum: 0,
-            identifier: 0xBABE,
-            sequence: seq,
-            payload: time.num_milliseconds() as u64,
-        };
-
-        let mut bytes = hdr.to_bytes();
-        let buf: &mut [u8] = bytes.as_mut_slice();
-
-        hdr.checksum = PingerICMPEcho::calculate_checksum(buf);
-        let bytes = hdr.to_bytes();
-        let buf = bytes.as_slice();
-
-        let addr: Box<dyn SockaddrLike>;
-
-        match reflector.is_ipv4() {
-            true => {
-                let ip4 = Ipv4Addr::from_str(&*reflector.to_string()).unwrap();
-                let sock4 = SocketAddrV4::new(ip4, 0);
-                addr = Box::new(SockaddrIn::from(sock4));
-            }
-            false => {
-                let ip6 = Ipv6Addr::from_str(&*reflector.to_string()).unwrap();
-                let sock6 = SocketAddrV6::new(ip6, 0, 0, 0);
-                addr = Box::new(SockaddrIn6::from(sock6));
-            }
-        }
-
-        sendto(self.socket, buf, addr.as_ref(), MsgFlags::empty()).expect("TODO: F message");
-
-        Ok(())
-    }
+    fn new(id: u16, reflectors: Vec<IpAddr>) -> Self;
+    fn receive_loop(&mut self);
+    fn sender_loop(&mut self);
+    fn send_ping(&mut self, reflector: &IpAddr, id: u16, seq: u16) -> Result<(), Box<dyn Error>>;
 }

src/pinger_icmp.rs

@@ -0,0 +1,124 @@
+use std::error::Error;
+use std::net::{IpAddr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
+use std::{process, thread, time};
+
+use crate::Pinger;
+use byteorder::*;
+use nix::sys::time::TimeValLike;
+use nix::time::{clock_gettime, ClockId};
+use pnet::packet::icmp::IcmpType;
+use pnet::packet::icmp::IcmpTypes::EchoReply;
+use pnet::packet::ip::IpNextHeaderProtocols;
+use pnet::packet::*;
+use pnet_transport::TransportChannelType::Layer4;
+use pnet_transport::TransportProtocol::Ipv4;
+use pnet_transport::*;
+
+pub struct PingerICMPEcho {
+    id: u16,
+    reflectors: Vec<IpAddr>,
+    rx: TransportReceiver,
+    tx: TransportSender,
+}
+
+impl Pinger for PingerICMPEcho {
+    fn new(id: u16, reflectors: Vec<IpAddr>) -> Self {
+        let protocol = Layer4(Ipv4(IpNextHeaderProtocols::Icmp));
+
+        // Create a new transport channel, dealing with layer 4 packets on a test protocol
+        // It has a receive buffer of 4096 bytes.
+        let (tx, rx) = match transport_channel(4096, protocol) {
+            Ok((tx, rx)) => (tx, rx),
+            Err(e) => panic!(
+                "An error occurred when creating the transport channel: {}",
+                e
+            ),
+        };
+
+        PingerICMPEcho {
+            id,
+            reflectors,
+            rx,
+            tx,
+        }
+    }
+
+    fn receive_loop(&mut self) {
+        let mut iter = icmp_packet_iter(&mut self.rx);
+        loop {
+            let (packet, sender) = match iter.next() {
+                Ok(res) => res,
+                Err(_) => continue,
+            };
+
+            let curr_time: u64 = clock_gettime(ClockId::CLOCK_MONOTONIC)
+                .unwrap()
+                .num_milliseconds() as u64;
+
+            if packet.get_icmp_type() != EchoReply {
+                continue;
+            }
+
+            let icmp_response = icmp::echo_reply::EchoReplyPacket::new(packet.packet()).unwrap();
+
+            if icmp_response.get_identifier() != self.id {
+                continue;
+            }
+
+            let sent_time: u64 = icmp_response.payload().read_u64::<NativeEndian>().unwrap();
+            let rtt = curr_time - sent_time;
+
+            println!("Type: {:4}  | Reflector IP: {:>15}  | Seq: {:5}  | Current time: {:8}  |  Sent time: {:8}  | RTT: {:8}", "ICMP", sender.to_string(), icmp_response.get_sequence_number(), curr_time, sent_time, rtt);
+        }
+    }
+
+    fn sender_loop(&mut self) {
+        println!("My ID is: {}", self.id);
+
+        let mut seq: u16 = 0;
+        let tick_duration_ms: u16 = 500;
+        let sleep_duration = time::Duration::from_millis(tick_duration_ms as u64);
+
+        loop {
+            let reflectors = self.reflectors.clone();
+
+            for reflector in reflectors.iter() {
+                self.send_ping(reflector, self.id, seq)
+                    .expect("TODO: panic message");
+            }
+
+            thread::sleep(sleep_duration);
+
+            seq += 1;
+
+            if seq > u16::MAX {
+                seq = 0;
+            }
+        }
+    }
+
+    fn send_ping(&mut self, reflector: &IpAddr, id: u16, seq: u16) -> Result<(), Box<dyn Error>> {
+        let time = clock_gettime(ClockId::CLOCK_MONOTONIC).unwrap();
+
+        let mut buf = [0u8; 8 + 56];
+
+        let mut packet = icmp::echo_request::MutableEchoRequestPacket::new(&mut buf).unwrap();
+
+        packet.populate(&icmp::echo_request::EchoRequest {
+            icmp_type: icmp::IcmpTypes::EchoRequest,
+            icmp_code: icmp::IcmpCode::new(0),
+            checksum: 0,
+            identifier: id,
+            sequence_number: seq,
+            payload: Vec::from(time.num_milliseconds().to_ne_bytes()),
+        });
+
+        packet.set_checksum(icmp::checksum(
+            &icmp::IcmpPacket::new(&packet.packet()).unwrap(),
+        ));
+
+        self.tx.send_to(packet, *reflector).expect("Error sending");
+
+        Ok(())
+    }
+}

src/pinger_icmp_ts.rs

@@ -0,0 +1,153 @@
+use bincode::{deserialize, serialize};
+use std::error::Error;
+use std::net::{IpAddr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
+use std::{process, thread, time};
+
+use crate::Pinger;
+use byteorder::*;
+use nix::sys::time::TimeValLike;
+use nix::time::{clock_gettime, ClockId};
+use pnet::packet::icmp::IcmpTypes::{Timestamp, TimestampReply};
+use pnet::packet::ip::IpNextHeaderProtocols;
+use pnet::packet::*;
+use pnet_transport::TransportChannelType::Layer4;
+use pnet_transport::TransportProtocol::Ipv4;
+use pnet_transport::*;
+use serde::{Deserialize, Serialize};
+
+#[derive(Deserialize, Serialize, Copy, Clone, Default, Debug)]
+pub struct ICMPTimestamp {
+    type_: u8,
+    code: u8,
+    checksum: u16,
+    identifier: u16,
+    sequence: u16,
+    originate_time: u32,
+    receive_time: u32,
+    transmit_time: u32,
+}
+
+pub struct PingerICMPTimestamps {
+    id: u16,
+    reflectors: Vec<IpAddr>,
+    rx: TransportReceiver,
+    tx: TransportSender,
+}
+
+impl Pinger for PingerICMPTimestamps {
+    fn new(id: u16, reflectors: Vec<IpAddr>) -> Self {
+        let protocol = Layer4(Ipv4(IpNextHeaderProtocols::Icmp));
+
+        // Create a new transport channel, dealing with layer 4 packets on a test protocol
+        // It has a receive buffer of 4096 bytes.
+        let (tx, rx) = match transport_channel(4096, protocol) {
+            Ok((tx, rx)) => (tx, rx),
+            Err(e) => panic!(
+                "An error occurred when creating the transport channel: {}",
+                e
+            ),
+        };
+
+        PingerICMPTimestamps {
+            id,
+            reflectors,
+            rx,
+            tx,
+        }
+    }
+
+    fn receive_loop(&mut self) {
+        let mut iter = icmp_packet_iter(&mut self.rx);
+        loop {
+            let (packet, sender) = match iter.next() {
+                Ok(res) => res,
+                Err(_) => continue,
+            };
+
+            let time = clock_gettime(ClockId::CLOCK_REALTIME).unwrap();
+            let time_since_midnight: i64 =
+                ((time.tv_sec() % 86400 * 1000) + (time.tv_nsec() / 1000000));
+
+            if packet.get_icmp_type() != TimestampReply {
+                continue;
+            }
+
+            let hdr: ICMPTimestamp = deserialize(packet.packet()).unwrap();
+
+            if hdr.identifier != self.id {
+                println!("wrong id: {} != {}", hdr.identifier, self.id);
+                continue;
+            }
+
+            let rtt: i64 = time_since_midnight - hdr.originate_time as i64;
+            let dl_time: i64 = time_since_midnight - hdr.transmit_time as i64;
+            let ul_time: i64 = hdr.receive_time as i64 - hdr.originate_time as i64;
+
+            println!("Type: {:4}  | Reflector IP: {:>15}  | Seq: {:5}  | Current time: {:8}  |  Originate: {:8}  |  Received time: {:8}  |  Transmit time : {:8}  |  RTT: {:8}  | UL time: {:5}  | DL time: {:5}", "ICMP", sender.to_string(), hdr.sequence, time_since_midnight, hdr.originate_time, hdr.receive_time, hdr.transmit_time, rtt, ul_time, dl_time);
+        }
+    }
+
+    fn sender_loop(&mut self) {
+        println!("My ID is: {}", self.id);
+
+        let mut seq: u16 = 0;
+        let tick_duration_ms: u16 = 500;
+        let sleep_duration = time::Duration::from_millis(tick_duration_ms as u64);
+
+        loop {
+            let reflectors = self.reflectors.clone();
+
+            for reflector in reflectors.iter() {
+                self.send_ping(reflector, self.id, seq)
+                    .expect("TODO: panic message");
+            }
+
+            thread::sleep(sleep_duration);
+
+            seq += 1;
+
+            if seq > u16::MAX {
+                seq = 0;
+            }
+        }
+    }
+
+    fn send_ping(&mut self, reflector: &IpAddr, id: u16, seq: u16) -> Result<(), Box<dyn Error>> {
+        let mut buf = [0u8; 8 + 56];
+
+        let mut packet = pnet::packet::icmp::MutableIcmpPacket::new(&mut buf).unwrap();
+
+        let time = clock_gettime(ClockId::CLOCK_REALTIME).unwrap();
+        let time_since_midnight: u32 =
+            ((time.tv_sec() % 86400 * 1000) + (time.tv_nsec() / 1000000)) as u32;
+
+        let mut icmp = ICMPTimestamp {
+            type_: 255,
+            code: 255,
+            checksum: 65535,
+            identifier: id,
+            sequence: seq,
+            originate_time: time_since_midnight,
+            receive_time: 0,
+            transmit_time: 0,
+        };
+
+        let buf_v: Vec<u8> = serialize(&icmp).unwrap();
+        let payload = &buf_v[4..];
+
+        packet.populate(&icmp::Icmp {
+            icmp_type: Timestamp,
+            icmp_code: icmp::IcmpCode::new(0),
+            checksum: 0,
+            payload: Vec::from(payload),
+        });
+
+        packet.set_checksum(icmp::checksum(
+            &icmp::IcmpPacket::new(&packet.packet()).unwrap(),
+        ));
+
+        self.tx.send_to(packet, *reflector).expect("Error sending");
+
+        Ok(())
+    }
+}