Fix TCP Kernel Sockets for DB Writer (#3)

* Add first tcp_sock_tracer

* First working tcp kernel sockets

* Fix IP read

* Add arrayref to cargo

* Add db/Cargo.lock to gitignore

* Working concept

* Add tcp_sock tracing fields

* Fix struct (de-) serialization for files

* Add sock_trace_entry to db writer

* Fix byte alignment errors for trace entries

* Add port filter mechanism

* Speedup UI refresh rate

* Fix endianness of dport

* Add missing HandlerConstraint

---------

Co-authored-by: Moritz Flüchter <moritz.fluechter@uni-tuebingen.de>

Author: MoritzFlu

db/Cargo.toml

@@ -18,6 +18,7 @@ indicatif = "0.17.11"
 env_logger = "0.11.6"
 aya-ebpf = "0.1.1"
 arrayref = "0.3.9"
+bincode = "1.3.3"
 
 [net]
 net.git-fetch-with-cli= true 

db/src/bindings/sock.rs

@@ -1,18 +1,18 @@
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 
+use serde::Deserialize;
 use ts_storage::{DataValue, IpTuple};
 
 use crate::{db_writer::DBOperation, flow_tracker::{EventIndexer, AF_INET}, reader::FromBuffer};
 
 use arrayref::array_ref;
-
 #[repr(C)]
-#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, Default)]
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, Default, Deserialize)]
 pub struct sock_trace_entry {
     pub time: u64,
     pub addr_v4: u64,
-    pub src_v6: [u8; 16],
-    pub dst_v6: [u8; 16],
+    pub src_v6: [u8; 16usize],
+    pub dst_v6: [u8; 16usize],
     pub ports: u32,
     pub family: u16,
     // SOCK Stats
@@ -45,29 +45,113 @@ pub struct sock_trace_entry {
     // TCP_SOCK -> tcp_options_received
     pub snd_wscale: u16,
     pub rcv_wscale: u16,
-    pub div: u32
+    pub div: [u8; 4usize],
 }
 
 impl FromBuffer for sock_trace_entry {
     fn from_buffer(buf: &Vec<u8>) -> Self {
-        unsafe { *(buf.as_ptr() as *const sock_trace_entry) }
+        //unsafe { *(buf.as_ptr() as *const sock_trace_entry) }
+
+        let try_deserialize = bincode::deserialize::<'_, sock_trace_entry>(buf);
+
+        if try_deserialize.is_err() {
+            sock_trace_entry::default()
+        } else {
+            try_deserialize.unwrap()
+        }
 
     }
+    const ENTRY_SIZE: usize = 160;
 }
 
 impl EventIndexer for sock_trace_entry {
     fn get_field(&self, index: usize) -> Option<DataValue> {
         match index {
+            0 => if self.pacing_rate > 0 {Some(DataValue::Int(self.pacing_rate as i64))} else {None},
+            1 => if self.max_pacing_rate > 0 {Some(DataValue::Int(self.max_pacing_rate as i64))} else {None},
+            2 => if self.backoff > 0 {Some(DataValue::Int(self.backoff as i64))} else {None},
+            3 => if self.rto > 0 {Some(DataValue::Int(self.rto as i64))} else {None},
+            4 => if self.ato > 0 {Some(DataValue::Int(self.ato as i64))} else {None},
+            5 => if self.rcv_mss > 0 {Some(DataValue::Int(self.rcv_mss as i64))} else {None},
+            6 => if self.snd_cwnd > 0 {Some(DataValue::Int(self.snd_cwnd as i64))} else {None},
+            7 => if self.bytes_acked > 0 {Some(DataValue::Int(self.bytes_acked as i64))} else {None},
+            8 => if self.snd_ssthresh > 0 {Some(DataValue::Int(self.snd_ssthresh as i64))} else {None},
+            9 => if self.total_retrans > 0 {Some(DataValue::Int(self.total_retrans as i64))} else {None},
+            10 => if self.probes > 0 {Some(DataValue::Int(self.probes as i64))} else {None},
+            11 => if self.lost > 0 {Some(DataValue::Int(self.lost as i64))} else {None},
+            12 => if self.sacked_out > 0 {Some(DataValue::Int(self.sacked_out as i64))} else {None},
+            13 => if self.retrans > 0 {Some(DataValue::Int(self.retrans as i64))} else {None},
+            14 => if self.rcv_ssthresh > 0 {Some(DataValue::Int(self.rcv_ssthresh as i64))} else {None},
+            15 => if self.rttvar > 0 {Some(DataValue::Int(self.rttvar as i64))} else {None},
+            16 => if self.advmss > 0 {Some(DataValue::Int(self.advmss as i64))} else {None},
+            17 => if self.reordering > 0 {Some(DataValue::Int(self.reordering as i64))} else {None},
+            18 => if self.rcv_rtt > 0 {Some(DataValue::Int(self.rcv_rtt as i64))} else {None},
+            19 => if self.rcv_space > 0 {Some(DataValue::Int(self.rcv_space as i64))} else {None},
+            20 => if self.bytes_received > 0 {Some(DataValue::Int(self.bytes_received as i64))} else {None},
+            21 => if self.segs_out> 0 {Some(DataValue::Int(self.segs_out as i64))} else {None},
+            22 => if self.segs_in > 0 {Some(DataValue::Int(self.segs_in as i64))} else {None},
+            23 => if self.snd_wscale > 0 {Some(DataValue::Int(self.snd_wscale as i64))} else {None},
+            24 => if self.rcv_wscale > 0 {Some(DataValue::Int(self.rcv_wscale as i64))} else {None},
             _ => None, // TODO: better error handling
         }
     }
     fn get_default_field(&self, index: usize) -> DataValue {
         match index {
+            0 => DataValue::Int(0),
+            1 => DataValue::Int(0),
+            2 => DataValue::Int(0),
+            3 => DataValue::Int(0),
+            4 => DataValue::Int(0),
+            5 => DataValue::Int(0),
+            6 => DataValue::Int(0),
+            7 => DataValue::Int(0),
+            8 => DataValue::Int(0),
+            9 => DataValue::Int(0),
+            10 => DataValue::Int(0),
+            11 => DataValue::Int(0),
+            12 => DataValue::Int(0),
+            13 => DataValue::Int(0),
+            14 => DataValue::Int(0),
+            15 => DataValue::Int(0),
+            16 => DataValue::Int(0),
+            17 => DataValue::Int(0),
+            18 => DataValue::Int(0),
+            19 => DataValue::Int(0),
+            20 => DataValue::Int(0),
+            21 => DataValue::Int(0),
+            22 => DataValue::Int(0),
+            23 => DataValue::Int(0),
+            24 => DataValue::Int(0),
             _ => panic!("Tried to access out of bounds index!"), // TODO: better error handling
         }
     }
     fn get_field_name(&self, index: usize) -> &str {
         match index {
+            0 => "pacing_rate",
+            1 => "max_pacing_rate",
+            2 => "backoff",
+            3 => "rto",
+            4 => "ato",
+            5 => "rcv_mss",
+            6 => "snd_cwnd",
+            7 => "bytes_acked",
+            8 => "snd_ssthresh",
+            9 => "total_retrans",
+            10 => "probes",
+            11 => "lost",
+            12 => "sacked_out",
+            13 => "retrans",
+            14 => "rcv_ssthresh",
+            15 => "rttvar",
+            16 => "advmss",
+            17 => "reordering",
+            18 => "rcv_rtt",
+            19 => "rcv_space",
+            20 => "bytes_received",
+            21 => "segs_out",
+            22 => "segs_in",
+            23 => "snd_wscale",
+            24 => "rcv_wscale",
             _ => panic!("Tried to access out of bounds index!"), // TODO: better error handling
         }
     }
@@ -114,12 +198,15 @@ impl EventIndexer for sock_trace_entry {
         }
     }
     fn get_max_index(&self) -> usize {
-        0
+        24
     }
     fn get_timestamp(&self) -> f64 {
         self.time as f64
     }
     fn as_db_op(self) -> DBOperation {
         DBOperation::Socket(self)
     }
+    fn get_struct_length(&self) -> usize {
+        160
+    }
 }

db/src/bindings/tcp_packet.rs

@@ -1,11 +1,12 @@
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 
+use serde::Deserialize;
 use ts_storage::{DataValue, IpTuple};
 
 use crate::{db_writer::DBOperation, flow_tracker::EventIndexer, reader::FromBuffer};
 
 #[repr(C)]
-#[derive(Debug, Clone, Copy, Default)]
+#[derive(Debug, Clone, Copy, Default, Deserialize)]
 pub struct TcpPacket {
     pub time: u64,
     pub saddr: u32,
@@ -24,14 +25,21 @@ pub struct TcpPacket {
     pub flag_syn: bool,
     pub flag_fin: bool,
     pub checksum: u16,
-    pub div: u32,
+    pub div: [u8; 4usize],
 }
 
 impl FromBuffer for TcpPacket {
     fn from_buffer(buf: &Vec<u8>) -> Self {
-        unsafe { *(buf.as_ptr() as *const TcpPacket) }
+        let try_deserialize = bincode::deserialize::<'_, TcpPacket>(buf);
+
+        if try_deserialize.is_err() {
+            TcpPacket::default()
+        } else {
+            try_deserialize.unwrap()
+        }
 
     }
+    const ENTRY_SIZE: usize = 74;
 }
 
 impl EventIndexer for TcpPacket {
@@ -110,4 +118,7 @@ impl EventIndexer for TcpPacket {
     fn as_db_op(self) -> DBOperation {
         DBOperation::Packet(self)
     }
+    fn get_struct_length(&self) -> usize {
+        74
+    }
 }

db/src/bindings/tcp_probe.rs

@@ -1,13 +1,14 @@
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 
+use serde::Deserialize;
 use ts_storage::{DataValue, IpTuple};
 
 use crate::{
     db_writer::DBOperation, flow_tracker::{EventIndexer, AF_INET}, reader::FromBuffer, shorten_to_ipv4, shorten_to_ipv6
 };
 
 #[repr(C)]
-#[derive(Debug, Clone, Copy, Default)]
+#[derive(Debug, Clone, Copy, Default, Deserialize)]
 pub struct TcpProbe {
     pub time: u64,
     pub saddr: [u8; 28usize],
@@ -25,13 +26,20 @@ pub struct TcpProbe {
     pub srtt: u32,
     pub rcv_wnd: u32,
     pub sock_cookie: u64,
-    pub div: u32,
+    pub div: [u8; 4usize],
 }
 
 impl FromBuffer for TcpProbe {
     fn from_buffer(buf: &Vec<u8>) -> Self {
-        unsafe { *(buf.as_ptr() as *const TcpProbe) }
+        let try_deserialize = bincode::deserialize::<'_, TcpProbe>(buf);
+
+        if try_deserialize.is_err() {
+            TcpProbe::default()
+        } else {
+            try_deserialize.unwrap()
+        }
     }
+    const ENTRY_SIZE: usize = 116;
 }
 
 impl EventIndexer for TcpProbe {
@@ -109,4 +117,7 @@ impl EventIndexer for TcpProbe {
     fn as_db_op(self) -> DBOperation {
         DBOperation::Probe(self)
     }
+    fn get_struct_length(&self) -> usize {
+        116
+    }
 }