[feature] Integration Test Passed

    1.Receive packets through pcap
    2.Packet decapsulate
    3.Trigger events
    4.Dispatch event management
    5.Call plugin handle
    6.Plugin manipulate session

3 files changed, 211 insertions, 15 deletions

diff --git a/src/packet/capture.rs b/src/packet/capture.rs
index aa1204d..fbbf6d7 100644
--- a/src/packet/capture.rs
+++ b/src/packet/capture.rs
@@ -1,4 +1,7 @@
+use crate::thread::thread::ThreadContex;
 use pcap::Capture;
+use std::cell::RefCell;
+use std::rc::Rc;
 
 pub struct PacketCapture {
     capture: Capture<pcap::Active>,
@@ -7,7 +10,7 @@ pub struct PacketCapture {
 impl PacketCapture {
     pub fn new(device: &str) -> Self {
         println!("Packet Capture Open Device {:?}", device);
-        let mut capture = Capture::from_device(device)
+        let capture = Capture::from_device(device)
             .unwrap()
             .immediate_mode(true)
             .open()
@@ -16,13 +19,18 @@ impl PacketCapture {
         PacketCapture { capture }
     }
 
-    pub fn poll_packet(&mut self, callback: fn(data: &[u8], len: u32)) {
+    pub fn poll_packet(
+        &mut self,
+        callback: fn(data: &[u8], len: u32, ctx: Rc<RefCell<ThreadContex>>),
+        ctx: Rc<RefCell<ThreadContex>>,
+    ) {
         let mut packet_num = 0;
         while let Ok(packet) = self.capture.next_packet() {
             packet_num += 1;
+            println!("\n==================== New Packet ====================\n");
             println!("Packet[{}]->header : {:?}", packet_num, packet.header);
             println!("Packet[{}]->data   : {:?}", packet_num, packet.data);
-            callback(&packet.data, packet.header.len);
+            callback(&packet.data, packet.header.len, ctx.clone());
         }
     }
 
diff --git a/src/packet/error.rs b/src/packet/error.rs
index 20eceec..fe0620e 100644
--- a/src/packet/error.rs
+++ b/src/packet/error.rs
@@ -45,4 +45,4 @@ impl core::fmt::Display for PacketError {
             PacketError::UnsupportAppProtocol => write!(f, "Unsupport App Protocol"),
         }
     }
-}
\ No newline at end of file
+}
diff --git a/src/packet/packet.rs b/src/packet/packet.rs
index 40403cb..8b2d460 100644
--- a/src/packet/packet.rs
+++ b/src/packet/packet.rs
@@ -1,3 +1,4 @@
+use crate::event::event::BuildInEvent;
 use crate::packet::error::PacketError;
 use crate::protocol::codec::Decode;
 use crate::protocol::dns::DNS_MESSAGE;
@@ -9,6 +10,13 @@ use crate::protocol::ipv4::IPv4Header;
 use crate::protocol::ipv6::IPv6Header;
 use crate::protocol::tcp::TcpHeader;
 use crate::protocol::udp::UdpHeader;
+use crate::session::session::packet2session;
+use crate::session::session::Session;
+use crate::session::session::SessionProto;
+use crate::session::session::SessionState;
+use crate::thread::thread::ThreadContex;
+use std::cell::RefCell;
+use std::rc::Rc;
 
 #[allow(non_camel_case_types)]
 #[derive(Clone, Debug, PartialEq)]
@@ -43,11 +51,11 @@ impl Packet<'_> {
         }
     }
 
-    pub fn handle(&mut self) -> Result<(), PacketError> {
+    pub fn handle(&mut self, ctx: Rc<RefCell<ThreadContex>>) -> Result<(), PacketError> {
         if self.orig_data.len() != self.orig_len as usize {
             return Err(PacketError::InvalidPacketLength);
         }
-        return handle_l2(self, self.orig_data);
+        return handle_l2(self, self.orig_data, ctx);
     }
 
     pub fn get_outer_l3_layer(&self) -> Option<Encapsulation> {
@@ -301,17 +309,117 @@ impl Packet<'_> {
 
         return None;
     }
+
+    pub fn get_trace_id(&self) -> Option<String> {
+        let num = self.encapsulation.len();
+        let mut trace_id = String::new();
+        for i in 0..num {
+            match self.encapsulation[i] {
+                Encapsulation::L3_IP4(ref l3_header, _) => {
+                    trace_id.push_str("IP4->IP4;");
+                    trace_id.push_str(&l3_header.source_address.to_string());
+                    trace_id.push_str("->");
+                    trace_id.push_str(&l3_header.dest_address.to_string());
+                    trace_id.push_str(";");
+                }
+                Encapsulation::L3_IP6(ref l3_header, _) => {
+                    trace_id.push_str("IP6->IP6;");
+                    trace_id.push_str(&l3_header.source_address.to_string());
+                    trace_id.push_str("->");
+                    trace_id.push_str(&l3_header.dest_address.to_string());
+                    trace_id.push_str(";");
+                }
+                Encapsulation::L4_TCP(ref l4_header, _) => {
+                    trace_id.push_str("TCP->TCP;");
+                    trace_id.push_str(&l4_header.source_port.to_string());
+                    trace_id.push_str("->");
+                    trace_id.push_str(&l4_header.dest_port.to_string());
+                    trace_id.push_str(";");
+                }
+                Encapsulation::L4_UDP(ref l4_header, _) => {
+                    trace_id.push_str("UDP->UDP;");
+                    trace_id.push_str(&l4_header.source_port.to_string());
+                    trace_id.push_str("->");
+                    trace_id.push_str(&l4_header.dest_port.to_string());
+                    trace_id.push_str(";");
+                }
+                _ => continue,
+            }
+        }
+
+        Some(trace_id)
+    }
+
+    pub fn get_reversed_trace_id(&self) -> Option<String> {
+        let num = self.encapsulation.len();
+        let mut trace_id = String::new();
+        for i in 0..num {
+            match self.encapsulation[i] {
+                Encapsulation::L3_IP4(ref l3_header, _) => {
+                    trace_id.push_str("IP4->IP4;");
+                    trace_id.push_str(&l3_header.dest_address.to_string());
+                    trace_id.push_str("->");
+                    trace_id.push_str(&l3_header.source_address.to_string());
+                    trace_id.push_str(";");
+                }
+                Encapsulation::L3_IP6(ref l3_header, _) => {
+                    trace_id.push_str("IP6->IP6;");
+                    trace_id.push_str(&l3_header.dest_address.to_string());
+                    trace_id.push_str("->");
+                    trace_id.push_str(&l3_header.source_address.to_string());
+                    trace_id.push_str(";");
+                }
+                Encapsulation::L4_TCP(ref l4_header, _) => {
+                    trace_id.push_str("TCP->TCP;");
+                    trace_id.push_str(&l4_header.dest_port.to_string());
+                    trace_id.push_str("->");
+                    trace_id.push_str(&l4_header.source_port.to_string());
+                    trace_id.push_str(";");
+                }
+                Encapsulation::L4_UDP(ref l4_header, _) => {
+                    trace_id.push_str("UDP->UDP;");
+                    trace_id.push_str(&l4_header.dest_port.to_string());
+                    trace_id.push_str("->");
+                    trace_id.push_str(&l4_header.source_port.to_string());
+                    trace_id.push_str(";");
+                }
+                _ => continue,
+            }
+        }
+
+        Some(trace_id)
+    }
 }
 
-fn handle_l2<'a>(packet: &mut Packet<'a>, input: &'a [u8]) -> Result<(), PacketError> {
+pub fn trace_id_reversed(trace_id: &String) -> String {
+    let mut reversed_trace_id = String::new();
+    let mut trace_id_vec: Vec<&str> = trace_id.split(";").collect();
+    trace_id_vec.pop();
+    for item in trace_id_vec.iter() {
+        let mut item_vec: Vec<&str> = item.split("->").collect();
+        item_vec.reverse();
+        reversed_trace_id.push_str(&item_vec.join("->"));
+        reversed_trace_id.push_str(";");
+    }
+    reversed_trace_id
+}
+
+fn handle_l2<'a>(
+    packet: &mut Packet<'a>,
+    input: &'a [u8],
+    ctx: Rc<RefCell<ThreadContex>>,
+) -> Result<(), PacketError> {
+    let event_mgr = ctx.borrow().get_event_mgr();
     let result = EthernetFrame::decode(input);
     if let Ok((payload, header)) = result {
         dbg!(&header);
         packet
             .encapsulation
             .push(Encapsulation::L2_ETH(header, payload));
-        return handle_l3(packet, payload, header.ether_type);
+        BuildInEvent::trigger_l2_event(event_mgr, None);
+        return handle_l3(packet, payload, header.ether_type, ctx);
     } else {
+        BuildInEvent::trigger_l2_event(event_mgr, None);
         return Err(PacketError::IncompleteEthernetFrame);
     }
 }
@@ -320,7 +428,9 @@ fn handle_l3<'a>(
     packet: &mut Packet<'a>,
     input: &'a [u8],
     next_proto: EtherType,
+    ctx: Rc<RefCell<ThreadContex>>,
 ) -> Result<(), PacketError> {
+    let event_mgr = ctx.borrow().get_event_mgr();
     match next_proto {
         EtherType::IPv4 => {
             let result = IPv4Header::decode(input);
@@ -337,8 +447,11 @@ fn handle_l3<'a>(
 
                 // TODO IPv4 Fragment
 
-                return handle_l4(packet, payload, header.protocol);
+                BuildInEvent::trigger_l3_event(event_mgr.clone(), None);
+                BuildInEvent::trigger_ip4_event(event_mgr, None);
+                return handle_l4(packet, payload, header.protocol, ctx);
             } else {
+                BuildInEvent::trigger_l3_event(event_mgr, None);
                 return Err(PacketError::IncompleteIpv4Header);
             }
         }
@@ -357,12 +470,16 @@ fn handle_l3<'a>(
 
                 // TODO IPv6 Fragment
 
-                return handle_l4(packet, payload, header.next_header);
+                BuildInEvent::trigger_l3_event(event_mgr.clone(), None);
+                BuildInEvent::trigger_ip6_event(event_mgr, None);
+                return handle_l4(packet, payload, header.next_header, ctx);
             } else {
+                BuildInEvent::trigger_l3_event(event_mgr, None);
                 return Err(PacketError::IncompleteIpv6Header);
             }
         }
         _e => {
+            BuildInEvent::trigger_l3_event(event_mgr, None);
             return Err(PacketError::UnsupportEthernetType);
         }
     }
@@ -372,7 +489,10 @@ fn handle_l4<'a>(
     packet: &mut Packet<'a>,
     input: &'a [u8],
     next_proto: IPProtocol,
+    ctx: Rc<RefCell<ThreadContex>>,
 ) -> Result<(), PacketError> {
+    let event_mgr = ctx.borrow().get_event_mgr();
+    let session_mgr = ctx.borrow().get_session_mgr();
     match next_proto {
         IPProtocol::UDP => {
             let result = UdpHeader::decode(input);
@@ -381,8 +501,24 @@ fn handle_l4<'a>(
                 packet
                     .encapsulation
                     .push(Encapsulation::L4_UDP(header, payload));
-                return handle_l7(packet, payload);
+
+                let session = packet2session(&packet, session_mgr);
+                BuildInEvent::trigger_l4_event(event_mgr.clone(), Some(session.clone()));
+                session.borrow_mut().set_session_proto(SessionProto::UDP);
+                match session.borrow().get_session_state() {
+                    SessionState::New => {
+                        BuildInEvent::trigger_udp_opening_event(event_mgr, Some(session.clone()));
+                    }
+                    SessionState::Active => {
+                        BuildInEvent::trigger_udp_active_event(event_mgr, Some(session.clone()));
+                    }
+                    SessionState::Inactive | SessionState::Expired => {
+                        BuildInEvent::trigger_udp_expire_event(event_mgr, Some(session.clone()));
+                    }
+                }
+                return handle_l7(packet, payload, session, ctx);
             } else {
+                BuildInEvent::trigger_l4_event(event_mgr, None);
                 return Err(PacketError::IncompleteUdpHeader);
             }
         }
@@ -390,22 +526,57 @@ fn handle_l4<'a>(
             let result = TcpHeader::decode(input);
             if let Ok((payload, header)) = result {
                 dbg!(&header);
+
+                let tcp_need_closed = header.flag_rst || header.flag_fin;
                 packet
                     .encapsulation
                     .push(Encapsulation::L4_TCP(header, payload));
+
                 // TODO TCP Reassembly
-                return handle_l7(packet, payload);
+
+                let session = packet2session(&packet, session_mgr);
+                BuildInEvent::trigger_l4_event(event_mgr.clone(), Some(session.clone()));
+                if tcp_need_closed {
+                    session
+                        .borrow_mut()
+                        .set_session_state(SessionState::Inactive);
+                }
+                session.borrow_mut().set_session_proto(SessionProto::TCP);
+                match session.borrow().get_session_state() {
+                    SessionState::New => {
+                        BuildInEvent::trigger_tcp_opening_event(event_mgr, Some(session.clone()));
+                    }
+                    SessionState::Active => {
+                        BuildInEvent::trigger_tcp_active_event(event_mgr, Some(session.clone()));
+                    }
+                    SessionState::Expired => {
+                        BuildInEvent::trigger_tcp_expire_event(event_mgr, Some(session.clone()));
+                    }
+                    SessionState::Inactive => {
+                        BuildInEvent::trigger_tcp_closed_event(event_mgr, Some(session.clone()));
+                    }
+                }
+
+                return handle_l7(packet, payload, session, ctx);
             } else {
+                BuildInEvent::trigger_l4_event(event_mgr, None);
                 return Err(PacketError::IncompleteTcpHeader);
             }
         }
         _e => {
+            BuildInEvent::trigger_l4_event(event_mgr, None);
             return Err(PacketError::UnsupportIPProtocol);
         }
     }
 }
 
-fn handle_l7<'a>(packet: &mut Packet<'a>, input: &'a [u8]) -> Result<(), PacketError> {
+fn handle_l7<'a>(
+    packet: &mut Packet<'a>,
+    input: &'a [u8],
+    session: Rc<RefCell<Session>>,
+    ctx: Rc<RefCell<ThreadContex>>,
+) -> Result<(), PacketError> {
+    let event_mgr = ctx.borrow().get_event_mgr();
     let next_proto = l7_identify(packet);
     match next_proto {
         L7Protocol::DNS => {
@@ -415,8 +586,11 @@ fn handle_l7<'a>(packet: &mut Packet<'a>, input: &'a [u8]) -> Result<(), PacketE
                 packet
                     .encapsulation
                     .push(Encapsulation::L7_DNS(header, payload));
+                BuildInEvent::trigger_l7_event(event_mgr.clone(), Some(session.clone()));
+                BuildInEvent::trigger_dns_event(event_mgr, Some(session));
                 return Ok(());
             } else {
+                BuildInEvent::trigger_l7_event(event_mgr, Some(session));
                 return Err(PacketError::IncompleteAppHeader);
             }
         }
@@ -427,12 +601,16 @@ fn handle_l7<'a>(packet: &mut Packet<'a>, input: &'a [u8]) -> Result<(), PacketE
                 packet
                     .encapsulation
                     .push(Encapsulation::L7_HTTP(header, payload));
+                BuildInEvent::trigger_l7_event(event_mgr.clone(), Some(session.clone()));
+                BuildInEvent::trigger_http_event(event_mgr, Some(session));
                 return Ok(());
             } else {
+                BuildInEvent::trigger_l7_event(event_mgr, Some(session));
                 return Err(PacketError::IncompleteAppHeader);
             }
         }
         L7Protocol::Unsupported => {
+            BuildInEvent::trigger_l7_event(event_mgr, Some(session));
             return Err(PacketError::UnsupportAppProtocol);
         }
     }
@@ -470,13 +648,17 @@ fn l7_identify(packet: &Packet) -> L7Protocol {
 mod tests {
     use super::Encapsulation;
     use super::Packet;
+    use crate::packet::packet::trace_id_reversed;
     use crate::protocol::ip::IPProtocol;
     use crate::protocol::ipv4::IPv4Header;
     use crate::protocol::ipv6::IPv6Header;
     use crate::protocol::tcp::TcpHeader;
     use crate::protocol::udp::UdpHeader;
+    use crate::thread::thread::ThreadContex;
+    use std::cell::RefCell;
     use std::net::Ipv4Addr;
     use std::net::Ipv6Addr;
+    use std::rc::Rc;
 
     #[test]
     fn test_packet_handle1() {
@@ -574,8 +756,9 @@ mod tests {
             0x00, 0x01, 0x00, 0x01, /* DNS END */
         ];
 
+        let thread_ctx = Rc::new(RefCell::new(ThreadContex::new()));
         let mut packet = Packet::new(&bytes, bytes.len() as u32);
-        let result = packet.handle();
+        let result = packet.handle(thread_ctx);
 
         match result {
             Ok(v) => {
@@ -715,8 +898,9 @@ mod tests {
             0x2a, 0x0d, 0x0a, 0x0d, 0x0a, /* HTTP END */
         ];
 
+        let thread_ctx = Rc::new(RefCell::new(ThreadContex::new()));
         let mut packet = Packet::new(&bytes, bytes.len() as u32);
-        let result = packet.handle();
+        let result = packet.handle(thread_ctx);
 
         match result {
             Ok(v) => {
@@ -853,5 +1037,9 @@ mod tests {
             packet.get_inner_l4_layer(),
             Some(Encapsulation::L4_UDP(udp_hdr, b"4"))
         );
+
+        assert_eq!(packet.get_trace_id(), Some("IP4->IP4;192.168.0.101->121.14.154.93;TCP->TCP;50081->443;IP6->IP6;2409:8034:4025::50:a31->2409:8034:4040:5301::204;UDP->UDP;9993->9994;".to_string()));
+        assert_eq!(packet.get_reversed_trace_id() ,Some("IP4->IP4;121.14.154.93->192.168.0.101;TCP->TCP;443->50081;IP6->IP6;2409:8034:4040:5301::204->2409:8034:4025::50:a31;UDP->UDP;9994->9993;".to_string()));
+        assert_eq!(trace_id_reversed(&packet.get_trace_id().unwrap()), "IP4->IP4;121.14.154.93->192.168.0.101;TCP->TCP;443->50081;IP6->IP6;2409:8034:4040:5301::204->2409:8034:4025::50:a31;UDP->UDP;9994->9993;".to_string());
     }
 }