Fix ingress traffic is same as egress traffic, split ebpf to ingress and egress ebpf

Cargo.lock

@@ -1232,7 +1232,18 @@ dependencies = [
 ]
 
 [[package]]
-name = "net-guardia-ebpf"
+name = "net-guardia-egress-ebpf"
+version = "0.1.0"
+dependencies = [
+ "aya-ebpf",
+ "aya-log-ebpf",
+ "net-guardia-common",
+ "network-types",
+ "which",
+]
+
+[[package]]
+name = "net-guardia-ingress-ebpf"
 version = "0.1.0"
 dependencies = [
  "aya-ebpf",

Cargo.toml

@@ -1,6 +1,6 @@
 [workspace]
 resolver = "2"
-members = ["net-guardia", "net-guardia-common", "net-guardia-ebpf"]
+members = ["net-guardia", "net-guardia-common", "net-guardia-ingress-ebpf", "net-guardia-egress-ebpf"]
 default-members = ["net-guardia", "net-guardia-common"]
 
 [workspace.dependencies]
@@ -26,12 +26,16 @@ panic = "abort"
 
 [profile.release]
 panic = "abort"
-opt-level = 3
-lto = true
-strip = true
-debug = false
-overflow-checks = false
+#opt-level = 3
+#lto = true
+#strip = true
+#debug = false
+#overflow-checks = false
 
-[profile.release.package.net-guardia-ebpf]
+[profile.release.package.net-guardia-ingress-ebpf]
+debug = 2
+codegen-units = 1
+
+[profile.release.package.net-guardia-egress-ebpf]
 debug = 2
 codegen-units = 1

net-guardia-egress-ebpf/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "net-guardia-ebpf"
+name = "net-guardia-egress-ebpf"
 version = "0.1.0"
 edition = "2021"
 
@@ -14,5 +14,5 @@ network-types = "0.0.7"
 which = { workspace = true }
 
 [[bin]]
-name = "net-guardia"
+name = "net-guardia-egress"
 path = "src/main.rs"

net-guardia-egress-ebpf/src/action/mod.rs

@@ -0,0 +1 @@
+pub mod statistics;

net-guardia-egress-ebpf/src/action/statistics.rs

@@ -0,0 +1,80 @@
+use aya_ebpf::helpers::bpf_ktime_get_ns;
+use aya_ebpf::macros::map;
+use aya_ebpf::maps::LruHashMap;
+use net_guardia_common::model::event::{IPv4Event, IPv6Event};
+use net_guardia_common::model::flow_stats::EbpfFlowStats;
+use net_guardia_common::model::ip_address::{EbpfAddrPortV4, EbpfAddrPortV6};
+use net_guardia_common::MAX_STATS;
+
+#[map]
+static IPV4_DST_1MIN: LruHashMap<EbpfAddrPortV4, EbpfFlowStats> =
+    LruHashMap::with_max_entries(MAX_STATS, 0);
+#[map]
+static IPV4_DST_10MIN: LruHashMap<EbpfAddrPortV4, EbpfFlowStats> =
+    LruHashMap::with_max_entries(MAX_STATS, 0);
+#[map]
+static IPV4_DST_1HOUR: LruHashMap<EbpfAddrPortV4, EbpfFlowStats> =
+    LruHashMap::with_max_entries(MAX_STATS, 0);
+#[map]
+static IPV6_DST_1MIN: LruHashMap<EbpfAddrPortV6, EbpfFlowStats> =
+    LruHashMap::with_max_entries(MAX_STATS, 0);
+#[map]
+static IPV6_DST_10MIN: LruHashMap<EbpfAddrPortV6, EbpfFlowStats> =
+    LruHashMap::with_max_entries(MAX_STATS, 0);
+#[map]
+static IPV6_DST_1HOUR: LruHashMap<EbpfAddrPortV6, EbpfFlowStats> =
+    LruHashMap::with_max_entries(MAX_STATS, 0);
+
+pub fn ipv4_update_stats(event: &IPv4Event) {
+    unsafe {
+        let now = bpf_ktime_get_ns();
+        let source = [event.source_ip, event.source_port as u32];
+        ipv4_update_flow_stats(&IPV4_DST_1MIN, &source, event, now);
+        ipv4_update_flow_stats(&IPV4_DST_10MIN, &source, event, now);
+        ipv4_update_flow_stats(&IPV4_DST_1HOUR, &source, event, now);
+    }
+}
+
+pub fn ipv6_update_stats(event: &IPv6Event) {
+    unsafe {
+        let now = bpf_ktime_get_ns();
+        let source = [event.source_ip, event.source_port as u128];
+        ipv6_update_flow_status(&IPV6_DST_1MIN, &source, event, now);
+        ipv6_update_flow_status(&IPV6_DST_10MIN, &source, event, now);
+        ipv6_update_flow_status(&IPV6_DST_1HOUR, &source, event, now);
+    }
+}
+
+#[inline(always)]
+unsafe fn ipv4_update_flow_stats(
+    map: &LruHashMap<EbpfAddrPortV4, EbpfFlowStats>,
+    key: &EbpfAddrPortV4,
+    event: &IPv4Event,
+    now: u64,
+) {
+    if let Some(status) = map.get_ptr_mut(key) {
+        (*status)[0] += event.len as u64;
+        (*status)[1] += 1;
+        (*status)[2] = now;
+    } else {
+        let new_stats = [event.len as u64, 1, now];
+        let _ = map.insert(key, &new_stats, 0);
+    }
+}
+
+#[inline(always)]
+unsafe fn ipv6_update_flow_status(
+    map: &LruHashMap<EbpfAddrPortV6, EbpfFlowStats>,
+    key: &EbpfAddrPortV6,
+    event: &IPv6Event,
+    now: u64,
+) {
+    if let Some(status) = map.get_ptr_mut(key) {
+        (*status)[0] += event.len as u64;
+        (*status)[1] += 1;
+        (*status)[2] = now;
+    } else {
+        let new_stats = [event.len as u64, 1, now];
+        let _ = map.insert(key, &new_stats, 0);
+    }
+}

net-guardia-egress-ebpf/src/main.rs

@@ -0,0 +1,69 @@
+#![no_std]
+#![no_main]
+mod action;
+mod utils;
+
+use action::statistics;
+use aya_ebpf::macros::{map, xdp};
+use aya_ebpf::maps::{PerCpuArray, ProgramArray};
+use aya_ebpf::{bindings::xdp_action, programs::XdpContext};
+use aya_log_ebpf::error;
+use net_guardia_common::model::event::Event;
+use network_types::eth::EtherType;
+use utils::parsing;
+
+#[map]
+static PROGRAM_ARRAY: ProgramArray = ProgramArray::with_max_entries(8, 0);
+#[map]
+static PARSED_PACKET: PerCpuArray<Event> = PerCpuArray::with_max_entries(1, 0);
+
+#[xdp]
+pub fn net_guardia(ctx: XdpContext) -> u32 {
+    match unsafe { parsing(ctx) } {
+        Ok(ret) => ret,
+        Err(_) => xdp_action::XDP_PASS,
+    }
+}
+
+unsafe fn parsing(ctx: XdpContext) -> Result<u32, ()> {
+    let start = ctx.data();
+    let end = ctx.data_end();
+    let event = parsing::parse_packet(start, end)?;
+    let ptr = PARSED_PACKET.get_ptr_mut(0).ok_or(())?;
+    let parsed_packet = ptr.as_mut().ok_or(())?;
+    *parsed_packet = event;
+    if PROGRAM_ARRAY.tail_call(&ctx, 0).is_err() {
+        error!(&ctx, "Tail call failed");
+    }
+    Err(())
+}
+
+#[xdp]
+pub fn statistics(ctx: XdpContext) -> u32 {
+    match unsafe { try_statistics(ctx) } {
+        Ok(ret) => ret,
+        Err(_) => xdp_action::XDP_PASS,
+    }
+}
+
+unsafe fn try_statistics(_: XdpContext) -> Result<u32, ()> {
+    let ptr = PARSED_PACKET.get_ptr(0).ok_or(())?;
+    let parsed_packet = ptr.read();
+    match parsed_packet.eth_type {
+        EtherType::Ipv4 => {
+            let event = parsed_packet.into_ipv4_event();
+            statistics::ipv4_update_stats(&event);
+        }
+        EtherType::Ipv6 => {
+            let event = parsed_packet.into_ipv6_event();
+            statistics::ipv6_update_stats(&event);
+        }
+        _ => Err(())?,
+    }
+    Ok(xdp_action::XDP_PASS)
+}
+
+#[panic_handler]
+fn panic(_info: &core::panic::PanicInfo) -> ! {
+    unsafe { core::hint::unreachable_unchecked() }
+}

net-guardia-ingress-ebpf/.cargo/config.toml

@@ -0,0 +1,12 @@
+# We have this so that one doesn't need to manually pass
+# --target=bpfel-unknown-none -Z build-std=core when running cargo
+# check/build/doc etc.
+#
+# NB: this file gets loaded only if you run cargo from this directory, it's
+# ignored if you run from the workspace root. See
+# https://doc.rust-lang.org/cargo/reference/config.html#hierarchical-structure
+[build]
+target = ["bpfeb-unknown-none", "bpfel-unknown-none"]
+
+[unstable]
+build-std = ["core"]

net-guardia-ingress-ebpf/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "net-guardia-ingress-ebpf"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+net-guardia-common = { path = "../net-guardia-common" }
+
+aya-ebpf = { workspace = true }
+aya-log-ebpf = { workspace = true }
+network-types = "0.0.7"
+
+[build-dependencies]
+which = { workspace = true }
+
+[[bin]]
+name = "net-guardia-ingress"
+path = "src/main.rs"

net-guardia-ingress-ebpf/build.rs

@@ -0,0 +1,17 @@
+use which::which;
+
+/// Building this crate has an undeclared dependency on the `bpf-linker` binary. This would be
+/// better expressed by [artifact-dependencies][bindeps] but issues such as
+/// https://github.com/rust-lang/cargo/issues/12385 make their use impractical for the time being.
+///
+/// This file implements an imperfect solution: it causes cargo to rebuild the crate whenever the
+/// mtime of `which bpf-linker` changes. Note that possibility that a new bpf-linker is added to
+/// $PATH ahead of the one used as the cache key still exists. Solving this in the general case
+/// would require rebuild-if-changed-env=PATH *and* rebuild-if-changed={every-directory-in-PATH}
+/// which would likely mean far too much cache invalidation.
+///
+/// [bindeps]: https://doc.rust-lang.org/nightly/cargo/reference/unstable.html?highlight=feature#artifact-dependencies
+fn main() {
+    let bpf_linker = which("bpf-linker").unwrap();
+    println!("cargo:rerun-if-changed={}", bpf_linker.to_str().unwrap());
+}

net-guardia-ingress-ebpf/rust-toolchain.toml

@@ -0,0 +1,3 @@
+[toolchain]
+channel = "nightly"
+components = ["rust-src"]

net-guardia-ingress-ebpf/src/lib.rs

@@ -0,0 +1,3 @@
+#![no_std]
+
+// This file exists to enable the library target.

net-guardia-ingress-ebpf/src/utils/mod.rs

@@ -0,0 +1 @@
+pub mod parsing;

net-guardia-ingress-ebpf/src/utils/parsing.rs

@@ -0,0 +1,106 @@
+use aya_ebpf::helpers::bpf_ktime_get_ns;
+use net_guardia_common::model::event::Event;
+use network_types::{
+    eth::{EthHdr, EtherType},
+    ip::{IpProto, Ipv4Hdr, Ipv6Hdr},
+    tcp::TcpHdr,
+    udp::UdpHdr,
+};
+
+pub fn parse_packet(start: usize, end: usize) -> Result<Event, ()> {
+    if start + size_of::<EthHdr>() > end {
+        return Err(());
+    }
+    let eth = unsafe { &*(start as *const EthHdr) };
+    match eth.ether_type {
+        EtherType::Ipv4 => parse_ipv4_packet(start, end),
+        EtherType::Ipv6 => parse_ipv6_packet(start, end),
+        _ => Err(())
+    }
+}
+
+#[inline(always)]
+pub fn parse_ipv4_packet(start: usize, end: usize) -> Result<Event, ()> {
+    let mut offset = size_of::<EthHdr>();
+    if start + offset + size_of::<Ipv4Hdr>() > end {
+        return Err(());
+    }
+    let ipv4 = unsafe { &*((start + offset) as *const Ipv4Hdr) };
+    offset += size_of::<Ipv4Hdr>();
+
+    let protocol = ipv4.proto;
+    let source_ip = u32::from_be(ipv4.src_addr);
+    let destination_ip = u32::from_be(ipv4.dst_addr);
+
+    let (source_port, destination_port) = match protocol {
+        IpProto::Tcp => parse_tcp_port(start, end, offset)?,
+        IpProto::Udp => parse_udp_port(start, end, offset)?,
+        _ => return Err(()),
+    };
+
+    Ok(Event {
+        eth_type: EtherType::Ipv4,
+        protocol,
+        source_ip: source_ip as u128,
+        destination_ip: destination_ip as u128,
+        source_port,
+        destination_port,
+        len: (end - start) as u32,
+        timestamp: unsafe { bpf_ktime_get_ns() },
+    })
+}
+
+#[inline(always)]
+pub fn parse_ipv6_packet(start: usize, end: usize) -> Result<Event, ()> {
+    let mut offset = size_of::<EthHdr>();
+    if start + offset + size_of::<Ipv6Hdr>() > end {
+        return Err(());
+    }
+    let ipv6 = unsafe { &*((start + offset) as *const Ipv6Hdr) };
+    offset += size_of::<Ipv6Hdr>();
+
+    let protocol = ipv6.next_hdr;
+    let source_ip = u128::from_be_bytes(unsafe { ipv6.src_addr.in6_u.u6_addr8 });
+    let destination_ip = u128::from_be_bytes(unsafe { ipv6.dst_addr.in6_u.u6_addr8 });
+
+    let (source_port, destination_port) = match protocol {
+        IpProto::Tcp => parse_tcp_port(start, end, offset)?,
+        IpProto::Udp => parse_udp_port(start, end, offset)?,
+        _ => return Err(()),
+    };
+
+    Ok(Event {
+        eth_type: EtherType::Ipv6,
+        protocol,
+        source_ip,
+        destination_ip,
+        source_port,
+        destination_port,
+        len: (end - start) as u32,
+        timestamp: unsafe { bpf_ktime_get_ns() },
+    })
+}
+
+#[inline(always)]
+fn parse_tcp_port(start: usize, end: usize, offset: usize) -> Result<(u16, u16), ()> {
+    let tcp: *const TcpHdr = (start + offset) as *const TcpHdr;
+    if start + offset + size_of::<TcpHdr>() > end {
+        return Err(());
+    }
+    Ok((
+        u16::from_be(unsafe { (*tcp).source }),
+        u16::from_be(unsafe { (*tcp).dest }),
+    ))
+}
+
+#[inline(always)]
+fn parse_udp_port(start: usize, end: usize, offset: usize) -> Result<(u16, u16), ()> {
+    let udp: *const UdpHdr = (start + offset) as *const UdpHdr;
+    if start + offset + size_of::<UdpHdr>() > end {
+        return Err(());
+    }
+    Ok((
+        u16::from_be(unsafe { (*udp).source }),
+        u16::from_be(unsafe { (*udp).dest }),
+    ))
+}

net-guardia/build.rs

@@ -7,7 +7,13 @@ use std::{
 
 use cargo_metadata::{Artifact, CompilerMessage, Message, Metadata, MetadataCommand, Package, Target, TargetKind};
 
-/// This crate has a runtime dependency on artifacts produced by the `net-guardia-ebpf` crate.
+
+fn main() {
+    build_ingress_ebpf();
+    build_egress_ebpf();
+}
+
+/// This crate has a runtime dependency on artifacts produced by the `net-guardia-ingress-ebpf` crate.
 /// This would be better expressed as one or more [artifact-dependencies][bindeps] but issues such
 /// as:
 ///
@@ -18,11 +24,11 @@ use cargo_metadata::{Artifact, CompilerMessage, Message, Metadata, MetadataComma
 /// prevent their use for the time being.
 ///
 /// [bindeps]: https://doc.rust-lang.org/nightly/cargo/reference/unstable.html?highlight=feature#artifact-dependencies
-fn main() {
+fn build_ingress_ebpf() {
     let Metadata { packages, .. } = MetadataCommand::new().no_deps().exec().unwrap();
     let ebpf_package = packages
         .into_iter()
-        .find(|Package { name, .. }| name == "net-guardia-ebpf")
+        .find(|Package { name, .. }| name == "net-guardia-ingress-ebpf")
         .unwrap();
 
     let out_dir = env::var_os("OUT_DIR").unwrap();
@@ -48,9 +54,9 @@ fn main() {
         let Package { manifest_path, .. } = ebpf_package;
         let ebpf_dir = manifest_path.parent().unwrap();
 
-        // We have a build-dependency on `net-guardia-ebpf`, so cargo will automatically rebuild us
-        // if `net-guardia-ebpf`'s *library* target or any of its dependencies change. Since we
-        // depend on `net-guardia-ebpf`'s *binary* targets, that only gets us half of the way. This
+        // We have a build-dependency on `net-guardia-ingress-ebpf`, so cargo will automatically rebuild us
+        // if `net-guardia-ingress-ebpf`'s *library* target or any of its dependencies change. Since we
+        // depend on `net-guardia-ingress-ebpf`'s *binary* targets, that only gets us half of the way. This
         // stanza ensures cargo will rebuild us on changes to the binaries too, which gets us the
         // rest of the way.
         println!("cargo:rerun-if-changed={}", ebpf_dir.as_str());
@@ -76,7 +82,7 @@ fn main() {
         cmd.current_dir(ebpf_dir);
 
         // Workaround for https://github.com/rust-lang/cargo/issues/6412 where cargo flocks itself.
-        let ebpf_target_dir = out_dir.join("net-guardia-ebpf");
+        let ebpf_target_dir = out_dir.join("net-guardia-ingress-ebpf");
         cmd.arg("--target-dir").arg(&ebpf_target_dir);
 
         let mut child = cmd
@@ -146,3 +152,143 @@ fn main() {
         }
     }
 }
+
+/// This crate has a runtime dependency on artifacts produced by the `net-guardia-egress-ebpf` crate.
+/// This would be better expressed as one or more [artifact-dependencies][bindeps] but issues such
+/// as:
+///
+/// * https://github.com/rust-lang/cargo/issues/12374
+/// * https://github.com/rust-lang/cargo/issues/12375
+/// * https://github.com/rust-lang/cargo/issues/12385
+///
+/// prevent their use for the time being.
+///
+/// [bindeps]: https://doc.rust-lang.org/nightly/cargo/reference/unstable.html?highlight=feature#artifact-dependencies
+fn build_egress_ebpf() {
+    let Metadata { packages, .. } = MetadataCommand::new().no_deps().exec().unwrap();
+    let ebpf_package = packages
+        .into_iter()
+        .find(|Package { name, .. }| name == "net-guardia-egress-ebpf")
+        .unwrap();
+
+    let out_dir = env::var_os("OUT_DIR").unwrap();
+    let out_dir = PathBuf::from(out_dir);
+
+    let endian = env::var_os("CARGO_CFG_TARGET_ENDIAN").unwrap();
+    let target = if endian == "big" {
+        "bpfeb"
+    } else if endian == "little" {
+        "bpfel"
+    } else {
+        panic!("unsupported endian={:?}", endian)
+    };
+
+    // TODO(https://github.com/rust-lang/cargo/issues/4001): Make this `false` if we can determine
+    // we're in a check build.
+    let build_ebpf = true;
+    if build_ebpf {
+        let arch = env::var_os("CARGO_CFG_TARGET_ARCH").unwrap();
+
+        let target = format!("{target}-unknown-none");
+
+        let Package { manifest_path, .. } = ebpf_package;
+        let ebpf_dir = manifest_path.parent().unwrap();
+
+        // We have a build-dependency on `net-guardia-egress-ebpf`, so cargo will automatically rebuild us
+        // if `net-guardia-egress-ebpf`'s *library* target or any of its dependencies change. Since we
+        // depend on `net-guardia-egress-ebpf`'s *binary* targets, that only gets us half of the way. This
+        // stanza ensures cargo will rebuild us on changes to the binaries too, which gets us the
+        // rest of the way.
+        println!("cargo:rerun-if-changed={}", ebpf_dir.as_str());
+
+        let mut cmd = Command::new("cargo");
+        cmd.args([
+            "build",
+            "-Z",
+            "build-std=core",
+            "--bins",
+            "--message-format=json",
+            "--release",
+            "--target",
+            &target,
+        ]);
+
+        cmd.env("CARGO_CFG_BPF_TARGET_ARCH", arch);
+
+        // Workaround to make sure that the rust-toolchain.toml is respected.
+        for key in ["RUSTUP_TOOLCHAIN", "RUSTC", "RUSTC_WORKSPACE_WRAPPER"] {
+            cmd.env_remove(key);
+        }
+        cmd.current_dir(ebpf_dir);
+
+        // Workaround for https://github.com/rust-lang/cargo/issues/6412 where cargo flocks itself.
+        let ebpf_target_dir = out_dir.join("net-guardia-egress-ebpf");
+        cmd.arg("--target-dir").arg(&ebpf_target_dir);
+
+        let mut child = cmd
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .spawn()
+            .unwrap_or_else(|err| panic!("failed to spawn {cmd:?}: {err}"));
+        let Child { stdout, stderr, .. } = &mut child;
+
+        // Trampoline stdout to cargo warnings.
+        let stderr = stderr.take().unwrap();
+        let stderr = BufReader::new(stderr);
+        let stderr = std::thread::spawn(move || {
+            for line in stderr.lines() {
+                let line = line.unwrap();
+                println!("cargo:warning={line}");
+            }
+        });
+
+        let stdout = stdout.take().unwrap();
+        let stdout = BufReader::new(stdout);
+        let mut executables = Vec::new();
+        for message in Message::parse_stream(stdout) {
+            #[allow(clippy::collapsible_match)]
+            match message.expect("valid JSON") {
+                Message::CompilerArtifact(Artifact {
+                        executable,
+                        target: Target { name, .. },
+                        ..
+                    }) => {
+                    if let Some(executable) = executable {
+                        executables.push((name, executable.into_std_path_buf()));
+                    }
+                }
+                Message::CompilerMessage(CompilerMessage { message, .. }) => {
+                    for line in message.rendered.unwrap_or_default().split('\n') {
+                        println!("cargo:warning={line}");
+                    }
+                }
+                Message::TextLine(line) => {
+                    println!("cargo:warning={line}");
+                }
+                _ => {}
+            }
+        }
+
+        let status = child
+            .wait()
+            .unwrap_or_else(|err| panic!("failed to wait for {cmd:?}: {err}"));
+        assert_eq!(status.code(), Some(0), "{cmd:?} failed: {status:?}");
+
+        stderr.join().map_err(std::panic::resume_unwind).unwrap();
+
+        for (name, binary) in executables {
+            let dst = out_dir.join(name);
+            let _: u64 = fs::copy(&binary, &dst)
+                .unwrap_or_else(|err| panic!("failed to copy {binary:?} to {dst:?}: {err}"));
+        }
+    } else {
+        let Package { targets, .. } = ebpf_package;
+        for Target { name, kind, .. } in targets {
+            if *kind != [TargetKind::Bin] {
+                continue;
+            }
+            let dst = out_dir.join(name);
+            fs::write(&dst, []).unwrap_or_else(|err| panic!("failed to create {dst:?}: {err}"));
+        }
+    }
+}

net-guardia/src/core/control/access_control.rs

@@ -45,7 +45,7 @@ impl AccessControl {
     pub async fn initialize() -> anyhow::Result<()> {
         info!("{}", SystemEntry::Initializing);
         let mut system = System::instance_mut().await;
-        let ebpf = &mut system.ebpf;
+        let ebpf = &mut system.ingress_ebpf;
         let mut ipv4_maps = StdHashMap::new();
         let mut ipv6_maps = StdHashMap::new();
         for (key, (ipv4_name, ipv6_name)) in Self::MAP_CONFIGS {

net-guardia/src/core/control/service.rs

@@ -32,7 +32,7 @@ impl Service {
     pub async fn initialize() -> anyhow::Result<()> {
         info!("{}", SystemEntry::Initializing);
         let mut system = System::instance_mut().await;
-        let ebpf = &mut system.ebpf;
+        let ebpf = &mut system.ingress_ebpf;
         let mut service = Service {
             ipv4_http_service: AyaHashMap::try_from(ebpf.take_map("IPV4_HTTP_SERVICE").unwrap())?,
             ipv6_http_service: AyaHashMap::try_from(ebpf.take_map("IPV6_HTTP_SERVICE").unwrap())?,

net-guardia/src/core/statistics.rs

@@ -9,7 +9,7 @@ use aya::Pod;
 use net_guardia_common::model::flow_stats::EbpfFlowStats;
 use net_guardia_common::model::ip_address::{EbpfAddrPortV4, EbpfAddrPortV6};
 use std::collections::HashMap as StdHashMap;
-use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
+use std::net::{SocketAddrV4, SocketAddrV6};
 use std::sync::OnceLock;
 use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
 use tracing::info;
@@ -23,7 +23,7 @@ pub struct Statistics {
 }
 
 impl Statistics {
-    const MAP_CONFIGS: [((Direction, TimeType), (&'static str, &'static str)); 6] = [
+    const INGRESS_MAPS: [((Direction, TimeType), (&'static str, &'static str)); 3] = [
         (
             (Direction::Source, TimeType::_1Min),
             ("IPV4_SRC_1MIN", "IPV6_SRC_1MIN"),
@@ -36,6 +36,9 @@ impl Statistics {
             (Direction::Source, TimeType::_1Hour),
             ("IPV4_SRC_1HOUR", "IPV6_SRC_1HOUR"),
         ),
+    ];
+
+    const EGRESS_MAPS: [((Direction, TimeType), (&'static str, &'static str)); 3] = [
         (
             (Direction::Destination, TimeType::_1Min),
             ("IPV4_DST_1MIN", "IPV6_DST_1MIN"),
@@ -53,20 +56,35 @@ impl Statistics {
     pub async fn initialize() -> anyhow::Result<()> {
         info!("{}", SystemEntry::Initializing);
         let mut system = System::instance_mut().await;
-        let ebpf = &mut system.ebpf;
         let mut ipv4_maps = StdHashMap::new();
         let mut ipv6_maps = StdHashMap::new();
-        for (key, (ipv4_name, ipv6_name)) in Self::MAP_CONFIGS {
+        let ingress_ebpf = &mut system.ingress_ebpf;
+        for (key, (ipv4_name, ipv6_name)) in Self::INGRESS_MAPS {
             ipv4_maps.insert(
                 key,
                 FlowMap {
-                    map: AyaHashMap::try_from(ebpf.take_map(ipv4_name).unwrap())?,
+                    map: AyaHashMap::try_from(ingress_ebpf.take_map(ipv4_name).unwrap())?,
                 },
             );
             ipv6_maps.insert(
                 key,
                 FlowMap {
-                    map: AyaHashMap::try_from(ebpf.take_map(ipv6_name).unwrap())?,
+                    map: AyaHashMap::try_from(ingress_ebpf.take_map(ipv6_name).unwrap())?,
+                },
+            );
+        }
+        let egress_ebpf = &mut system.egress_ebpf;
+        for (key, (ipv4_name, ipv6_name)) in Self::EGRESS_MAPS {
+            ipv4_maps.insert(
+                key,
+                FlowMap {
+                    map: AyaHashMap::try_from(egress_ebpf.take_map(ipv4_name).unwrap())?,
+                },
+            );
+            ipv6_maps.insert(
+                key,
+                FlowMap {
+                    map: AyaHashMap::try_from(egress_ebpf.take_map(ipv6_name).unwrap())?,
                 },
             );
         }

net-guardia/src/core/system.rs

@@ -19,10 +19,13 @@ use tracing::{error, info, warn};
 static SYSTEM: OnceLock<RwLock<System>> = OnceLock::new();
 
 pub struct System {
-    pub ebpf: Ebpf,
+    pub ingress_ebpf: Ebpf,
+    pub egress_ebpf: Ebpf,
     pub boot_time: u64,
     #[allow(dead_code)]
-    program_array: ProgramArray<MapData>,
+    ingress_program_array: ProgramArray<MapData>,
+    #[allow(dead_code)]
+    egress_program_array: ProgramArray<MapData>,
 }
 
 impl System {
@@ -39,32 +42,50 @@ impl System {
 
     async fn ebpf_initialize() -> anyhow::Result<()> {
         let config = ConfigManager::now().await;
-        let interface = config.ingress_ifindex;
+        let ingress_interface = config.ingress_ifindex;
+        let egress_interface = config.egress_ifindex;
         let boot_time = SystemInfo::boot_time() * 1_000_000_000;
         Self::set_memory_limit()?;
-        let mut ebpf = Ebpf::load(aya::include_bytes_aligned!(concat!(
+        let mut ingress_ebpf = Ebpf::load(aya::include_bytes_aligned!(concat!(
             env!("OUT_DIR"),
-            "/net-guardia"
+            "/net-guardia-ingress"
         )))?;
-        if let Err(e) = aya_log::EbpfLogger::init(&mut ebpf) {
+        let mut egress_ebpf = Ebpf::load(aya::include_bytes_aligned!(concat!(
+            env!("OUT_DIR"),
+            "/net-guardia-egress"
+        )))?;
+        if let Err(e) = aya_log::EbpfLogger::init(&mut ingress_ebpf) {
             error!("{}", e);
             warn!("{}", EbpfEntry::LoggerInitializeFailed);
         }
-        let mut program_array = ProgramArray::try_from(ebpf.take_map("PROGRAM_ARRAY").unwrap())?;
-        Self::load_program(&mut ebpf, &mut program_array, "access_control", 0)?;
-        Self::load_program(&mut ebpf, &mut program_array, "service", 1)?;
-        // Self::load_program(&mut ebpf, &mut program_array, "defence", 2)?;
-        Self::load_program(&mut ebpf, &mut program_array, "sampling", 3)?;
-        Self::load_program(&mut ebpf, &mut program_array, "statistics", 4)?;
-        let program: &mut Xdp = ebpf.program_mut("net_guardia").unwrap().try_into()?;
-        program.load()?;
-        program
-            .attach(&interface, XdpFlags::default())
+        if let Err(e) = aya_log::EbpfLogger::init(&mut egress_ebpf) {
+            error!("{}", e);
+            warn!("{}", EbpfEntry::LoggerInitializeFailed);
+        }
+        let mut ingress_program_array = ProgramArray::try_from(ingress_ebpf.take_map("PROGRAM_ARRAY").unwrap())?;
+        Self::load_program(&mut ingress_ebpf, &mut ingress_program_array, "access_control", 0)?;
+        Self::load_program(&mut ingress_ebpf, &mut ingress_program_array, "service", 1)?;
+        // Self::load_program(&mut ebpf, &mut ingress_program_array, "defence", 2)?;
+        Self::load_program(&mut ingress_ebpf, &mut ingress_program_array, "sampling", 3)?;
+        Self::load_program(&mut ingress_ebpf, &mut ingress_program_array, "statistics", 4)?;
+        let ingress_program: &mut Xdp = ingress_ebpf.program_mut("net_guardia").unwrap().try_into()?;
+        let mut egress_program_array = ProgramArray::try_from(egress_ebpf.take_map("PROGRAM_ARRAY").unwrap())?;
+        Self::load_program(&mut egress_ebpf, &mut egress_program_array, "statistics", 0)?;
+        let egress_program: &mut Xdp = egress_ebpf.program_mut("net_guardia").unwrap().try_into()?;
+        ingress_program.load()?;
+        ingress_program
+            .attach(&ingress_interface, XdpFlags::default())
+            .context(EbpfEntry::AttachProgramFailed)?;
+        egress_program.load()?;
+        egress_program
+            .attach(&egress_interface, XdpFlags::default())
             .context(EbpfEntry::AttachProgramFailed)?;
         let system = System {
-            ebpf,
+            ingress_ebpf,
+            egress_ebpf,
             boot_time,
-            program_array,
+            ingress_program_array,
+            egress_program_array,
         };
         SYSTEM.get_or_init(|| RwLock::new(system));
         info!("{}", EbpfEntry::AttachProgramSuccess);