eguard: dns lpm match fix

Author: chriskaliX

plugins/eguard/config.yaml

@@ -27,6 +27,9 @@ dns:
   - name: "eguard_egress_test_dns_1"
     action: DENY
     domain: "*.baidu.com"
+  - name: "eguard_egress_test_dns_2"
+    action: DENY
+    domain: "*.loooooooooooooooooooooooooooooooooooog.com"
 file:
   - name: "file_test_1"
     action: DENY

plugins/eguard/src/bpf/eguard.skel.rs

@@ -7,9 +7,10 @@
 #include "common/general.h"
 #include "vmlinux.h"
 
-#define DNS_OFFSET    (sizeof(struct ethhdr) + sizeof(struct iphdr) + sizeof(struct udphdr))
-#define TYPE_DNS      3201
-#define MAX_DNS_NAME  255
+#define TYPE_DNS            3201
+#define DNS_OFFSET          (sizeof(struct ethhdr) + sizeof(struct iphdr) + sizeof(struct udphdr))
+#define DNS_MAX_LEN         256
+#define DNS_MAX_PRELEN      (DNS_MAX_LEN) * 8
 
 // The header contains the following fields:
 //     +--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+
@@ -38,7 +39,7 @@ struct dnshdr _dnshdr = {0};
 // DNS ACL structs
 struct dns_policy_key {
     __u32 prefixlen;
-    char  domain[255];
+    char  domain[DNS_MAX_LEN];
 };
 
 struct dns_policy_value {
@@ -58,8 +59,6 @@ struct {
 
 #define SIZE_DNSHDR         sizeof(struct dnshdr)
 #define SIZE_CONTEXT        sizeof(net_context_t)
-#define DNS_MAX_LEN         255
-#define DNS_MAX_READ_LEN    256
 #define MID_WAY             8192
 
 static __always_inline int tc_context_fill(net_packet_t pkt);
@@ -84,7 +83,7 @@ static __always_inline int l7_acl_rule(net_packet_t pkt, struct __sk_buff *skb)
 
         // trim key
         struct dns_policy_key key = {0};
-        key.prefixlen = 255;
+        key.prefixlen = DNS_MAX_PRELEN;
 
         bpf_skb_load_bytes(skb, DNS_OFFSET, (void *)&pkt.buf_p->buf[SIZE_CONTEXT], SIZE_DNSHDR);
         int offset = load_dns(pkt, &key, skb);

plugins/eguard/src/bpf/rules/l7_acl.h

@@ -16,22 +16,19 @@ pub struct DnsPolicy {
 impl CfgTrait for DnsPolicy {
     fn to_bytes(&self) -> Result<(Vec<u8>, Vec<u8>)> {
         let mut key = eguard_bss_types::dns_policy_key::default();
+        key.prefixlen = 256;
         let mut value = eguard_bss_types::dns_policy_value::default();
         let mut domain = self.domain.clone();
-        if domain.len() == 0 || domain.len() > 255 {
+        if domain.len() == 0 || domain.len() > 256 {
             bail!("domain length error: {}", domain.len());
         }
-
-        if domain.contains('*') {
+        if domain.starts_with('*') {
             domain.remove(0);
-            key.prefixlen = domain.len() as u32;
-        } else {
-            key.prefixlen = 255;
+            key.prefixlen = (domain.len() * 8) as u32;
         }
-
         let domain_bytes = domain.as_bytes();
-        let key_domain: [i8; 255] = {
-            let mut arr = [0; 255];
+        let key_domain: [i8; 256] = {
+            let mut arr = [0; 256];
             for (i, &byte) in domain_bytes.iter().enumerate() {
                 arr[domain.len() - i - 1] = byte as i8;
             }
@@ -43,7 +40,7 @@ impl CfgTrait for DnsPolicy {
             Action::LOG => 1,
         };
 
-        // flush to the map
+        // convert into bytes
         let key = unsafe { plain::as_bytes(&key) }.to_vec();
         let value = unsafe { plain::as_bytes(&value) }.to_vec();
         Ok((key, value))

plugins/eguard/src/config/parser/dns.rs

@@ -56,7 +56,7 @@ impl<'a> BpfProgram for DnsEvent {
         Ok(())
     }
 
-    fn detech(&mut self, _skel: &mut EguardSkel) -> Result<()> {
+    fn detach(&mut self, _skel: &mut EguardSkel) -> Result<()> {
         self.status.store(false, Ordering::SeqCst);
         Ok(())
     }

plugins/eguard/src/event/dns.rs

@@ -23,8 +23,8 @@ pub trait BpfProgram: Sync {
     /// attach bpf binary
     fn attach(&mut self, skel: &mut EguardSkel) -> Result<()>;
 
-    /// detech the binary, wrapper the destory method inside if it is needed
-    fn detech(&mut self, skel: &mut EguardSkel) -> Result<()>;
+    /// detach the binary, wrapper the destory method inside if it is needed
+    fn detach(&mut self, skel: &mut EguardSkel) -> Result<()>;
 
     /// status of the bpf program
     fn status(&self) -> bool;

plugins/eguard/src/event/mod.rs

@@ -101,13 +101,16 @@ impl<'a> BpfProgram for TcEvent {
         Ok(())
     }
 
-    /// actually, it is destory
-    fn detech(&mut self, skel: &mut EguardSkel) -> Result<()> {
-        let mut destroy_all = libbpf_rs::TcHook::new(skel.progs().hades_egress().as_fd());
-        destroy_all
-            .ifindex(self.if_idx)
-            .attach_point(TC_EGRESS | TC_INGRESS);
-        destroy_all.destroy()?;
+    /// actually, it is destroy
+    fn detach(&mut self, _skel: &mut EguardSkel) -> Result<()> {
+        if let Some(mut egress_hook) = self.egress_hook {
+            egress_hook.detach()?;
+            egress_hook.destroy()?;
+        }
+        if let Some(mut ingress_hook) = self.ingress_hook {
+            ingress_hook.detach()?;
+            ingress_hook.destroy()?;
+        }
         Ok(())
     }
 

plugins/eguard/src/event/tc.rs

@@ -10,7 +10,7 @@ use std::path::PathBuf;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 use std::thread;
-use std::time::Duration;
+use tokio::time::{timeout, Duration};
 
 use crate::config::config::Config as BpfConfig;
 use crate::event::dns::DnsEvent;
@@ -23,11 +23,6 @@ mod manager;
 pub const TYPE_TC: u32 = 3200;
 pub const TYPE_DNS: u32 = 3201;
 
-use std::alloc::System;
-
-#[global_allocator]
-static GLOBAL: System = System;
-
 fn main() -> Result<()> {
     let mut client = Client::new(false);
     set_boxed_logger(Box::new(Logger::new(Config {
@@ -47,6 +42,7 @@ fn main() -> Result<()> {
     let control_s = Arc::new(AtomicBool::new(false));
     let control_l = control_s.clone();
     let control_c = control_s.clone();
+    let control_r = control_s.clone();
     ctrlc::set_handler(move || {
         control_c.store(true, Ordering::SeqCst);
     })?;
@@ -77,37 +73,8 @@ fn main() -> Result<()> {
     }
 
     info!("init bpf program successfully");
-    // task_receive thread
     let mut client_c = client.clone();
-    let timeout = Duration::from_millis(500);
-    let _ = thread::Builder::new()
-        .name("task_receive".to_owned())
-        .spawn(move || loop {
-            match client_c.receive() {
-                Ok(task) => {
-                    let config = match serde_json::from_str::<BpfConfig>(task.get_data()) {
-                        Ok(config) => config,
-                        Err(e) => {
-                            error!("parse task failed: {}", e);
-                            continue;
-                        }
-                    };
-
-                    if let Err(e) = mgr_c.lock().unwrap().flush_config(config) {
-                        error!("flush task failed: {}", e);
-                        continue;
-                    }
 
-                    info!("task parse success")
-                }
-                Err(e) => {
-                    error!("when receiving task,an error occurred:{}", e);
-                    control_s.store(true, Ordering::Relaxed);
-                    return;
-                }
-            }
-        });
-    info!("task receive handler is running");
     // record_send thread
     let record_send = thread::Builder::new()
         .name("record_send".to_string())
@@ -116,7 +83,7 @@ fn main() -> Result<()> {
                 break;
             }
 
-            let rec = rx.recv_timeout(timeout);
+            let rec = rx.recv_timeout(std::time::Duration::from_millis(100));
             match rec {
                 Ok(rec) => {
                     if let Err(err) = client.send_record(&rec) {
@@ -130,6 +97,43 @@ fn main() -> Result<()> {
         })
         .unwrap();
     let _ = record_send.join();
+    // task_receive thread
+    let rt = tokio::runtime::Runtime::new()?;
+    rt.block_on(async {
+        info!("task receive handler is running");
+        loop {
+            if control_r.load(Ordering::SeqCst) {
+                break;
+            }
+            // receive data
+            let task = match timeout(Duration::from_millis(1000), client_c.receive_async()).await {
+                Ok(task) => match task {
+                    Ok(task) => task,
+                    Err(_) => {
+                        break;
+                    }
+                },
+                Err(_) => {
+                    continue;
+                }
+            };
+            // exit
+            let config = match serde_json::from_str::<BpfConfig>(task.get_data()) {
+                Ok(config) => config,
+                Err(e) => {
+                    error!("parse task failed: {}", e);
+                    continue;
+                }
+            };
+
+            if let Err(e) = mgr_c.lock().unwrap().flush_config(config) {
+                error!("flush task failed: {}", e);
+                continue;
+            }
+
+            info!("task parse success");
+        }
+    });
     info!("plugin will exit");
     Ok(())
 }

plugins/eguard/src/main.rs

@@ -3,7 +3,7 @@ mod eguard_skel {
 }
 use crate::{
     config::config::Config,
-    event::{eguard_skel::eguard_bss_types, event::TX, BpfProgram},
+    event::{eguard_skel::eguard_bss_types, BpfProgram},
     TYPE_TC,
 };
 use anyhow::{anyhow, bail, Ok, Result};
@@ -26,8 +26,6 @@ use std::{
 
 use crate::event::eguard_skel::{EguardSkel, EguardSkelBuilder};
 use byteorder::{LittleEndian, ReadBytesExt};
-use coarsetime::Clock;
-use sdk::{Payload, Record};
 use std::io::Cursor;
 
 lazy_static! {
@@ -67,21 +65,14 @@ impl Bpfmanager<'_> {
         // load the skel
         let mut skel = EguardSkelBuilder::default().open()?.load()?;
 
-        let network_perf = PerfBufferBuilder::new(skel.maps_mut().events())
+        let perf_buffer = PerfBufferBuilder::new(skel.maps_mut().events())
             .sample_cb(Bpfmanager::handle_tc_event)
             .lost_cb(Bpfmanager::handle_tc_lost_events)
             .build()?;
-        let exec_perf = PerfBufferBuilder::new(skel.maps_mut().exec_events())
-            .sample_cb(Bpfmanager::handle_exec_event)
-            .lost_cb(Bpfmanager::handle_exec_lost_events)
-            .build()?;
 
         let thread_handle = spawn(move || {
             while running.load(Ordering::SeqCst) {
-                if let Err(_) = network_perf.poll(Duration::from_millis(100)) {
-                    break;
-                }
-                if let Err(_) = exec_perf.poll(Duration::from_millis(100)) {
+                if let Err(_) = perf_buffer.poll(Duration::from_millis(100)) {
                     break;
                 }
             }
@@ -152,61 +143,17 @@ impl Bpfmanager<'_> {
     fn handle_tc_lost_events(cpu: i32, count: u64) {
         error!("lost tc {} events on CPU {}", count, cpu);
     }
-
-    /// working on this
-    fn handle_exec_event(_cpu: i32, data: &[u8]) {
-        // parse the context
-        let mut context = eguard_bss_types::data_context::default();
-        plain::copy_from_bytes(&mut context, data).expect("context decode failed");
-        let mut map = HashMap::new();
-        map.insert("cgroupid".to_string(), context.cgroup_id.to_string());
-        map.insert("pns".to_string(), context.pns.to_string());
-        map.insert("pid".to_string(), context.pid.to_string());
-        map.insert("tid".to_string(), context.tid.to_string());
-        map.insert("uid".to_string(), context.uid.to_string());
-        map.insert("gid".to_string(), context.gid.to_string());
-        map.insert("ppid".to_string(), context.ppid.to_string());
-        map.insert("pgid".to_string(), context.pgid.to_string());
-        map.insert("sessionid".to_string(), context.sessionid.to_string());
-        let comm: &[u8] = unsafe { std::mem::transmute(&context.comm[..]) };
-        map.insert("comm".to_string(), trim_null_chars(comm));
-        let pcomm: &[u8] = unsafe { std::mem::transmute(&context.pcomm[..]) };
-        map.insert("pcomm".to_string(), trim_null_chars(pcomm));
-        let nodename: &[u8] = unsafe { std::mem::transmute(&context.nodename[..]) };
-        map.insert("nodename".to_string(), trim_null_chars(nodename));
-
-        let mut rec = Record::new();
-        let mut pld = Payload::new();
-        pld.set_fields(map);
-        rec.set_timestamp(Clock::now_since_epoch().as_secs() as i64);
-        rec.set_data(pld);
-        rec.data_type = context.dt as i32;
-        let lock = TX
-            .lock()
-            .map_err(|e| error!("unable to acquire notification send channel: {}", e));
-        match &mut *lock.unwrap() {
-            Some(sender) => {
-                if let Err(err) = sender.send(rec) {
-                    error!("send failed: {}", err);
-                    return;
-                }
-            }
-            None => return,
-        }
-    }
-
-    fn handle_exec_lost_events(cpu: i32, count: u64) {
-        error!("lost exec_events {} events on CPU {}", count, cpu);
-    }
 }
 
 impl Drop for Bpfmanager<'_> {
     fn drop(&mut self) {
         let events = &mut *EVENTS.write().unwrap();
         for (key, e) in events.iter_mut() {
             if let Some(skel) = self.skel.as_mut() {
-                if let Err(err) = e.detech(skel) {
+                if let Err(err) = e.detach(skel) {
                     error!("drop event {} failed: {}", key, err);
+                } else {
+                    info!("drop event {} success", key);
                 }
             }
         }
@@ -215,18 +162,10 @@ impl Drop for Bpfmanager<'_> {
         if let Some(thread) = self.thread_handle.take() {
             thread.join().ok();
         }
-
         debug!("has dropped bpfmanager from thread");
     }
 }
 
-fn trim_null_chars(data: &[u8]) -> String {
-    String::from_utf8_lossy(data)
-        .to_string()
-        .trim_end_matches('\0')
-        .to_string()
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;