Wip: egress connections

nflux/Cargo.toml

@@ -17,6 +17,7 @@ serde = { version = "1.0.217", features = ["derive"] }
 toml = "0.8.19"
 bytes = "1.8.0"
 dns-lookup = "2.0.4"
+sysinfo = "0.33.1"
 
 [build-dependencies]
 cargo_metadata = { workspace = true }

nflux/src/egress.rs

@@ -9,7 +9,7 @@ use bytes::BytesMut;
 use tracing::{error, info, warn};
 use nflux_common::{convert_protocol, EgressConfig, EgressEvent};
 use crate::config::{Egress, IsEnabled};
-use crate::utils::lookup_address;
+use crate::utils::{get_process_name, lookup_address};
 
 pub fn populate_egress_config(bpf: &mut Ebpf, config: Egress) -> anyhow::Result<()> {
     let mut egress_config = Array::<_, EgressConfig>::try_from(
@@ -105,13 +105,14 @@ pub async fn process_egress_events(
             match parse_egress_event(buf) {
                 Ok(event) => {
                     info!(
-                        "program=tc_egress protocol={}, ip={}, src_port={}, dst_port={}, fqdn={}, pid={}",
+                        "program=tc_egress protocol={}, ip={}, src_port={}, dst_port={}, fqdn={}, pid={}, comm={}",
                         convert_protocol(event.protocol),
                         Ipv4Addr::from(event.dst_ip),
                         event.src_port,
                         event.dst_port,
                         lookup_address(event.dst_ip),
                         event.pid,
+                        get_process_name(event.pid)
                     );
                 }
                 Err(e) => error!("Failed to parse egress event on CPU {}: {}", cpu_id, e),

nflux/src/main.rs

@@ -9,6 +9,7 @@ use aya::maps::AsyncPerfEventArray;
 use aya::util::online_cpus;
 use aya::{include_bytes_aligned, Ebpf};
 use aya_log::EbpfLogger;
+use std::process;
 
 use config::{IsEnabled, Nflux};
 use egress::populate_egress_config;
@@ -34,6 +35,9 @@ async fn main() -> anyhow::Result<()> {
         std::process::exit(1);
     }
 
+    // Welcome message
+    info!("Starting nflux with pid {}", process::id());
+
     // Set memory limit
     set_mem_limit();
 

nflux/src/utils.rs

@@ -2,6 +2,7 @@ use std::{collections::HashMap, net::{IpAddr, Ipv4Addr, Ipv6Addr}};
 use dns_lookup::lookup_addr;
 use libc::getuid;
 use nflux_common::utils::is_private_ip;
+use sysinfo::{Pid, System};
 use tokio::signal;
 use tracing::{info, warn};
 
@@ -74,3 +75,16 @@ pub fn lookup_address(ip: u32) -> String {
         },
     }
 }
+
+pub fn get_process_name(pid: u64) -> String {
+    let mut s = System::new_all();
+
+    s.refresh_all();
+
+    match s.process(Pid::from(pid as usize)) {
+        Some(process) => {
+            format!("{:?}", process.name()).to_string()
+        }
+        None => String::new(),
+    }
+}