Working with TC egress implementation

nflux-ebpf/src/egress.rs

@@ -0,0 +1,36 @@
+use aya_ebpf::bindings::TC_ACT_PIPE;
+use aya_ebpf::macros::map;
+use aya_ebpf::maps::{LruHashMap, PerfEventArray};
+use aya_ebpf::programs::TcContext;
+use network_types::eth::{EthHdr, EtherType};
+use network_types::ip::Ipv4Hdr;
+use nflux_common::EgressEvent;
+
+#[map]
+static ACTIVE_CONNECTIONS: LruHashMap<u32, u32> = LruHashMap::with_max_entries(1024, 0);
+
+#[map]
+static EGRESS_EVENT: PerfEventArray<EgressEvent> = PerfEventArray::new(0);
+
+pub fn try_tc_egress(ctx: TcContext) -> Result<i32, ()> {
+    let ethhdr: EthHdr = ctx.load(0).map_err(|_| ())?;
+    match ethhdr.ether_type {
+        EtherType::Ipv4 => unsafe {
+            let ipv4hdr: Ipv4Hdr = ctx.load(EthHdr::LEN).map_err(|_| ())?;
+            let destination = u32::from_be(ipv4hdr.dst_addr);
+
+            // Check if this destination is already active
+            if ACTIVE_CONNECTIONS.get(&destination).is_none() {
+                // Log only new connections
+                let event = EgressEvent { dst_ip: destination };
+                EGRESS_EVENT.output(&ctx, &event, 0);
+
+                // Mark connection as active
+                ACTIVE_CONNECTIONS.insert(&destination, &1, 0).map_err(|_| ())?;
+            }
+        }
+        _ => return Ok(TC_ACT_PIPE),
+    }
+
+    Ok(TC_ACT_PIPE)
+}

nflux-ebpf/src/main.rs

@@ -2,6 +2,8 @@
 #![no_main]
 #![allow(nonstandard_style, dead_code)]
 
+mod egress;
+
 use aya_ebpf::maps::lpm_trie::Key;
 use aya_ebpf::maps::{Array, LpmTrie, LruHashMap};
 use aya_ebpf::{
@@ -22,6 +24,7 @@ use network_types::{
     udp::UdpHdr,
 };
 use nflux_common::{ConnectionEvent, EgressEvent, IpRule, LpmKeyIpv4, LpmKeyIpv6};
+use crate::egress::try_tc_egress;
 
 #[map]
 static IPV4_RULES: LpmTrie<LpmKeyIpv4, IpRule> = LpmTrie::with_max_entries(1024, 0);
@@ -35,12 +38,6 @@ static CONNECTION_EVENTS: PerfEventArray<ConnectionEvent> = PerfEventArray::new(
 #[map]
 static ICMP_RULE: Array<u32> = Array::with_max_entries(1, 0);
 
-#[map]
-static EGRESS_EVENT: PerfEventArray<EgressEvent> = PerfEventArray::new(0);
-
-#[map]
-static ACTIVE_CONNECTIONS: LruHashMap<u32, u32> = LruHashMap::with_max_entries(1024, 0);
-
 #[xdp]
 pub fn nflux(ctx: XdpContext) -> u32 {
     match start_nflux(ctx) {
@@ -78,30 +75,6 @@ fn log_new_connection(ctx: XdpContext, src_addr: u32, dst_port: u16, protocol: u
     CONNECTION_EVENTS.output(&ctx, &event, 0);
 }
 
-fn try_tc_egress(ctx: TcContext) -> Result<i32, ()> {
-    let ethhdr: EthHdr = ctx.load(0).map_err(|_| ())?;
-    match ethhdr.ether_type {
-        EtherType::Ipv4 => unsafe {
-            let ipv4hdr: Ipv4Hdr = ctx.load(EthHdr::LEN).map_err(|_| ())?;
-            let destination = u32::from_be(ipv4hdr.dst_addr);
-
-            // Check if this destination is already active
-            if ACTIVE_CONNECTIONS.get(&destination).is_none() {
-                // Log only new connections
-                let event = EgressEvent { dst_ip: destination };
-                EGRESS_EVENT.output(&ctx, &event, 0);
-
-                // Mark connection as active
-                ACTIVE_CONNECTIONS.insert(&destination, &1, 0).map_err(|_| ())?;
-            }
-        }
-        _ => return Ok(TC_ACT_PIPE),
-    }
-
-    Ok(TC_ACT_PIPE)
-}
-
-
 fn start_nflux(ctx: XdpContext) -> Result<u32, ()> {
     let ethhdr: *const EthHdr = unsafe { ptr_at(&ctx, 0)? };
 

nflux.toml

@@ -12,7 +12,7 @@ log_type = "text"  # text or json. Defaults to text if not set
 
 [ip_rules]
 # The /32 CIDR block is used to represent a single IP address rather than a range
-"192.168.0.0/24" = { priority = 1, action = "allow", ports = [22, 8000], protocol = "tcp", log = false, description = "Allow SSH for entire local net" }
+"192.168.0.0/24" = { priority = 1, action = "allow", ports = [22, 8000, 80], protocol = "tcp", log = false, description = "Allow SSH for entire local net" }
 
 # curl -6 -v http://\[::ffff:192.168.0.26\]:80
 "fe80::5bc2:662b:ac2f:7e8b/128" = { priority = 2, action = "allow", ports = [80], protocol = "tcp", log = false, description = "Deny HTTP for specific IPv6 address" }

nflux/Cargo.toml

@@ -20,6 +20,7 @@ toml = "0.8.19"
 prometheus = "0.13.4"
 axum = "0.7.7"
 bytes = "1.8.0"
+dns-lookup = "2.0.4"
 
 [build-dependencies]
 cargo_metadata = { workspace = true }

nflux/src/main.rs

@@ -23,6 +23,7 @@ use tokio::task;
 use tracing::{error, info};
 use utils::{is_root_user, wait_for_shutdown};
 use crate::ebpf_mapping::populate_icmp_rule;
+use crate::utils::lookup_address;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {

nflux/src/utils.rs

@@ -1,3 +1,5 @@
+use std::net::{IpAddr, Ipv4Addr};
+use dns_lookup::lookup_addr;
 use libc::getuid;
 use tokio::signal;
 use tracing::{info, warn};
@@ -14,3 +16,14 @@ pub async fn wait_for_shutdown() -> anyhow::Result<()> {
     warn!("Exiting...");
     Ok(())
 }
+
+pub fn lookup_address(ip: u32) -> String {
+    // Convert the u32 IP address to Ipv4Addr
+    let ip = Ipv4Addr::from(ip);
+
+    // Convert to IpAddr for compatibility with lookup_addr
+    let ip = IpAddr::V4(ip);
+
+    // Perform the reverse DNS lookup
+    lookup_addr(&ip).unwrap_or_else(|_| "Unknown host".to_string())
+}