Logging pid for every egress packet

nflux-common/src/lib.rs

@@ -5,6 +5,13 @@ pub const MAX_RULES_PORT: usize = 32;
 pub const MAX_ALLOWED_PORTS: usize = 1024;
 pub const MAX_ALLOWED_IPV4: usize = 1024;
 
+#[repr(C)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct EgressConfig {
+    pub log_udp_connections: u8, // 0 = no, 1 = yes
+    pub log_tcp_connections: u8, // 0 = no, 1 = yes
+}
+
 #[repr(C)]
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct ConnectionEvent {
@@ -21,6 +28,7 @@ pub struct EgressEvent {
     pub src_port: u16,
     pub dst_port: u16,
     pub protocol: u8,
+    pub pid: u64,
 }
 
 #[repr(C)]
@@ -53,6 +61,7 @@ pub mod user {
     unsafe impl aya::Pod for IpRule {}
     unsafe impl aya::Pod for LpmKeyIpv4 {}
     unsafe impl aya::Pod for LpmKeyIpv6 {}
+    unsafe impl aya::Pod for EgressConfig {}
 }
 
 // Define the default configuration if the user does not provide one

nflux-ebpf/src/egress.rs

@@ -1,6 +1,7 @@
 use core::mem;
 
 use aya_ebpf::bindings::TC_ACT_PIPE;
+use aya_ebpf::helpers::gen::bpf_get_current_pid_tgid;
 use aya_ebpf::programs::TcContext;
 use aya_log_ebpf::info;
 use network_types::eth::{EthHdr, EtherType};
@@ -9,7 +10,7 @@ use network_types::tcp::TcpHdr;
 use network_types::udp::UdpHdr;
 use nflux_common::EgressEvent;
 
-use crate::maps::{ACTIVE_CONNECTIONS, EGRESS_EVENT};
+use crate::maps::{ACTIVE_CONNECTIONS, EGRESS_CONFIG, EGRESS_EVENT};
 
 
 #[inline]
@@ -25,10 +26,18 @@ fn ptr_at<T>(ctx: &TcContext, offset: usize) -> Result<*const T, ()> {
     Ok((start + offset) as *const T)
 }
 
-
 pub fn try_tc_egress(ctx: TcContext) -> Result<i32, ()> {
     let ethhdr: EthHdr = ctx.load(0).map_err(|_| ())?;
 
+    let egress_config = EGRESS_CONFIG.get(0).ok_or(())?;
+
+    // if let Some(&egress_config) = EGRESS_CONFIG.get(0) {
+    //     let egress_config = egres
+    //     if egress_config.log_udp == 0 {
+    //         info!(&ctx, "log_udp is disabled");
+    //     }
+    // }
+
     match ethhdr.ether_type {
         EtherType::Ipv4 => unsafe {
             let ipv4hdr: Ipv4Hdr = ctx.load(EthHdr::LEN).map_err(|_| ())?;
@@ -40,11 +49,13 @@ pub fn try_tc_egress(ctx: TcContext) -> Result<i32, ()> {
                     let src_port = u16::from_be((*tcphdr).source);
                     let dst_port = u16::from_be((*tcphdr).dest);
                     let protocol = IpProto::Tcp as u8;
+                    let pid_tgid = bpf_get_current_pid_tgid();
+                    let pid = pid_tgid >> 32;
 
                     // Check if this destination is already active
                     if ACTIVE_CONNECTIONS.get(&destination).is_none() {
                         // Log only new connections
-                        let event = EgressEvent { dst_ip: destination, src_port, dst_port, protocol };
+                        let event = EgressEvent { dst_ip: destination, src_port, dst_port, protocol, pid };
                         EGRESS_EVENT.output(&ctx, &event, 0);
 
                         // Mark connection as active
@@ -59,8 +70,10 @@ pub fn try_tc_egress(ctx: TcContext) -> Result<i32, ()> {
                     let src_port = u16::from_be((*udphdr).source);
                     let dst_port = u16::from_be((*udphdr).dest);
                     let protocol = IpProto::Udp as u8;
+                    let pid_tgid = bpf_get_current_pid_tgid();
+                    let pid = pid_tgid >> 32;
 
-                    let event = EgressEvent { dst_ip: destination, src_port, dst_port, protocol };
+                    let event = EgressEvent { dst_ip: destination, src_port, dst_port, protocol, pid};
                     EGRESS_EVENT.output(&ctx, &event, 0);
 
                     return Ok(TC_ACT_PIPE)
@@ -76,7 +89,7 @@ pub fn try_tc_egress(ctx: TcContext) -> Result<i32, ()> {
 
             // Create here a fake event
             // IPV6 is not implemented yet
-            let event = EgressEvent { dst_ip: u32::from_be_bytes([192, 67, 4, 2]), src_port: 111, dst_port: 99, protocol: 6 };
+            let event = EgressEvent { dst_ip: u32::from_be_bytes([192, 67, 4, 2]), src_port: 111, dst_port: 99, protocol: 6, pid: 1234};
             EGRESS_EVENT.output(&ctx, &event, 0);
             return Ok(TC_ACT_PIPE)
         }

nflux-ebpf/src/maps.rs

@@ -1,5 +1,5 @@
 use aya_ebpf::{macros::map, maps::{Array, LpmTrie, LruHashMap, PerfEventArray}};
-use nflux_common::{ConnectionEvent, EgressEvent, IpRule, LpmKeyIpv4, LpmKeyIpv6};
+use nflux_common::{ConnectionEvent, EgressConfig, EgressEvent, IpRule, LpmKeyIpv4, LpmKeyIpv6};
 
 #[map]
 pub static IPV4_RULES: LpmTrie<LpmKeyIpv4, IpRule> = LpmTrie::with_max_entries(1024, 0);
@@ -10,6 +10,9 @@ pub static IPV6_RULES: LpmTrie<LpmKeyIpv6, IpRule> = LpmTrie::with_max_entries(1
 #[map]
 pub static ICMP_RULE: Array<u32> = Array::with_max_entries(1, 0);
 
+#[map]
+pub static EGRESS_CONFIG: Array<EgressConfig> = Array::with_max_entries(1, 0);
+
 #[map]
 pub static FIREWALL_EVENTS: PerfEventArray<ConnectionEvent> = PerfEventArray::new(0);
 

nflux.toml

@@ -17,7 +17,9 @@ icmp_ping = "true" # Allow or deny ICMP ping requests
 [egress]
 # By the moment, working with physical interfaces (not virtual, like VPNs)
 enabled = "true"
-interfaces = ["wlo1"]
+interfaces = ["enp0s20f0u4"]
+log_udp_connections = "false" # Do not log UDP packets
+loc_tcp_connections = "true" # Log TCP packets
 log_private_connections = "false" # Do not show private connections in the logs
 
 #[egress_rules]

nflux/src/config.rs

@@ -40,6 +40,8 @@ pub struct Firewall {
 pub struct Egress {
     pub enabled: IsEnabled,
     pub interfaces: Vec<String>,
+    pub log_udp_connections: IsEnabled,
+    pub log_tcp_connections: IsEnabled,
     #[allow(dead_code)]
     pub log_private_connections: IsEnabled,
 }

nflux/src/egress.rs

@@ -80,23 +80,25 @@ pub async fn process_egress_events(
                     match log_private_connections {
                         IsEnabled::True => {
                              info!(
-                                "program=tc_egress protocol={}, ip={}, src_port={}, dst_port={}, fqdn={}",
+                                "program=tc_egress protocol={}, ip={}, src_port={}, dst_port={}, fqdn={}, pid={}",
                                 convert_protocol(event.protocol),
                                 Ipv4Addr::from(event.dst_ip),
                                 event.src_port,
                                 event.dst_port,
                                 "Private IP",
+                                event.pid,
                             );
                         }
                         IsEnabled::False => {
                             if ! is_private_ip(event.dst_ip) {
                              info!(
-                                "program=tc_egress protocol={}, ip={}, src_port={}, dst_port={}, fqdn={}",
+                                "program=tc_egress protocol={}, ip={}, src_port={}, dst_port={}, fqdn={}, pid={}",
                                 convert_protocol(event.protocol),
                                 Ipv4Addr::from(event.dst_ip),
                                 event.src_port,
                                 event.dst_port,
                                 lookup_address(event.dst_ip),
+                                event.pid,
                             );
                         }
                         }

nflux/src/firewall.rs

@@ -15,7 +15,7 @@ use crate::utils::{parse_cidr_v4, parse_cidr_v6};
 
 pub fn populate_icmp_rule(bpf: &mut Ebpf, icmp_ping: IsEnabled) -> anyhow::Result<()> {
     let mut settings_map = Array::<_, u32>::try_from(
-        bpf.map_mut("ICMP_RULE").context("Failed to find GLOBAL_SETTINGS map")?,
+        bpf.map_mut("ICMP_RULE").context("Failed to find ICMP_RULE map")?,
     )?;
 
     let value = match icmp_ping {

nflux/src/main.rs

@@ -5,7 +5,7 @@ mod firewall;
 mod egress;
 
 use anyhow::Context;
-use aya::maps::AsyncPerfEventArray;
+use aya::maps::{Array, AsyncPerfEventArray};
 use aya::util::online_cpus;
 use aya::{include_bytes_aligned, Ebpf};
 use aya_log::EbpfLogger;
@@ -14,6 +14,7 @@ use config::{IsEnabled, Nflux};
 use firewall::{attach_xdp_program, process_firewall_events};
 use log::warn;
 use logger::setup_logger;
+use nflux_common::EgressConfig;
 use tokio::task;
 use tracing::{error, info};
 use utils::{is_root_user, print_firewall_rules, set_mem_limit, wait_for_shutdown};
@@ -61,6 +62,25 @@ async fn main() -> anyhow::Result<()> {
     match config.egress.enabled {
         IsEnabled::True => {
             attach_tc_egress_program(&mut bpf, &config.egress.interfaces)?;
+            let mut egress_config = Array::<_, EgressConfig>::try_from(
+                    bpf.map_mut("EGRESS_CONFIG").context("Failed to find EGRESS_CONFIG map")?,
+                )?;
+
+                let config = EgressConfig {
+                    log_udp_connections: match config.egress.log_udp_connections {
+                        IsEnabled::True => 1,
+                        IsEnabled::False => 0,
+                    },
+                    log_tcp_connections: match config.egress.log_tcp_connections {
+                        IsEnabled::True => 1,
+                        IsEnabled::False => 0,
+                    },
+                };
+
+                egress_config
+                    .set(0, config, 0)
+                    .context("Failed to set ICMP_MAP")?;
+
             info!("TC egress started successfully!")
         }
         IsEnabled::False => {