Merge branch 'improve-update-performance' into 'main'

replace bpf hashmap with bpf array containing bloomfilter (as bitarray)

See merge request s1783831/rescan!1

.gitignore

@@ -13,6 +13,7 @@ Cargo.lock
 **/*.rs.bk
 
 # editor
+.projectile
 
 # Data files
 data/

responder-common/Cargo.toml

@@ -5,12 +5,10 @@ edition = "2021"
 
 [features]
 default = []
-user = [ "aya" ]
-
-[dependencies]
-# aya = { version = ">=0.11", optional=true }
-# aya = { path = "/home/niels/files/temp/aya/aya", optional=true }
-aya = { git = "https://github.com/aya-rs/aya", optional=true }
+user = []
 
 [lib]
 path = "src/lib.rs"
+
+[dependencies]
+konst = { version = "0.3", features = ["parsing"] }

responder-common/src/lib.rs

@@ -1 +1,56 @@
-#![no_std]
+#![no_std]
+
+pub mod bloom_filter {
+    use konst::primitive::{parse_usize, parse_u32};
+    use konst::unwrap_ctx;
+    use konst::option::unwrap_or;
+
+    pub const ADDRESS_BITS: usize = unwrap_ctx!(parse_usize(unwrap_or!(option_env!("BLOOMFILTER_ADDRESS_BITS"), "30"))); // default = 0x1E = 30
+    pub const BITS: usize = 1 << ADDRESS_BITS; // default = 0x40_000_000 = 128MB
+    pub const ADDRESS_MASK: usize = BITS - 1;
+
+    pub const WORD_BITS: usize = 0x8; // 8
+    pub type WordType = u8;
+    pub const ADDRESS_BITS_WORD: usize = 0x3; // log_2(0x8)
+    pub const ADDRESS_MASK_WORD: usize = WORD_BITS - 1; // 0x7
+
+    pub const ADDRESS_BITS_CHUNK: usize = unwrap_ctx!(parse_usize(unwrap_or!(option_env!("BLOOMFILTER_ADDRESS_BITS_CHUNK"), "18"))); // default = 0x12 = 18 = log_2(32KB)
+    pub const CHUNK_BITS: usize = 1 << ADDRESS_BITS_CHUNK; // default = 0x40000  = 32KB (per-cpu map value size limit)
+    pub const CHUNK_BYTES: usize = CHUNK_BITS >> 0x3;
+    pub const CHUNK_SIZE: usize = CHUNK_BITS >> ADDRESS_BITS_WORD; // WORD_SIZE * CHUNK_SIZE = 32Kb
+    pub type ChunkType = [WordType; CHUNK_SIZE];
+    pub const ADDRESS_MASK_CHUNK: usize = CHUNK_BITS - 1;
+
+    pub const BUFFER_SIZE: usize = BITS >> ADDRESS_BITS_WORD; // 0x20_0000
+    pub const MAP_SIZE: usize = BITS >> ADDRESS_BITS_CHUNK;
+
+    pub const HASH_COUNT: u32 = unwrap_ctx!(parse_u32(unwrap_or!(option_env!("BLOOMFILTER_HASH_COUNT"), "10")));
+    // Hashing based on jhash.h from kernel // TODO improve reference
+    const HASH_INITVAL: u32 = 0xdeadbeef;
+
+    #[inline(always)]
+    pub fn hash(key: u32, initial_value: u32) -> u32 {
+        let mut b = HASH_INITVAL.wrapping_add(initial_value);
+        let mut c = b;
+
+        let mut a = key.wrapping_add(b);
+
+        c ^= b;
+        c = c.wrapping_sub(b.rotate_left(14));
+        a ^= c;
+        a = a.wrapping_sub(c.rotate_left(11));
+        b ^= a;
+        b = b.wrapping_sub(a.rotate_left(25));
+        c ^= b;
+        c = c.wrapping_sub(b.rotate_left(16));
+        a ^= c;
+        a = a.wrapping_sub(c.rotate_left(4));
+        b ^= a;
+        b = b.wrapping_sub(a.rotate_left(14));
+        c ^= b;
+        c = c.wrapping_sub(b.rotate_left(24));
+
+        return c & ADDRESS_MASK as u32;
+    }
+}
+

responder-ebpf/Cargo.toml

@@ -4,8 +4,6 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-# aya-bpf = { path = "/home/niels/files/temp/aya/bpf/aya-bpf" }
-# aya-log-ebpf = { path = "/home/niels/files/temp/aya/bpf/aya-log-ebpf" }
 aya-bpf = { git = "https://github.com/aya-rs/aya" }
 aya-log-ebpf ={ git = "https://github.com/aya-rs/aya" }
 responder-common = { path = "../responder-common" }

responder-ebpf/src/bin/syn.rs

@@ -4,18 +4,21 @@
 use aya_bpf::{
     bindings::xdp_action,
     macros::{map, xdp},
-    maps::HashMap,
+    maps::Array,
     programs::XdpContext,
 };
 use aya_log_ebpf::info;
 
 use responder_ebpf::util::*;
 use responder_ebpf::bindings::tcphdr;
+use responder_common::*;
+
 use core::mem;
 
 const TCP_HDR_LEN: usize = mem::size_of::<tcphdr>();
 const IPPROTO_TCP: u8 = 0x06;
 
+
 #[inline(always)]
 fn parse_tcphdr(ctx: &XdpContext, cursor: &mut usize) -> Option<*mut tcphdr> {
     let tcp = ptr_at_mut::<tcphdr>(&ctx, *cursor);
@@ -44,18 +47,31 @@ unsafe fn bounce_tcp(_ctx: &XdpContext, tcp: *mut tcphdr) {
 }
 
 #[map(name = "FILTER_MAP")]
-static FILTER_MAP: HashMap<u32, u8> =
-    HashMap::<u32, u8>::with_max_entries(0x4000000, 0);
+static FILTER_MAP: Array<bloom_filter::ChunkType> =
+    Array::<bloom_filter::ChunkType>::with_max_entries(bloom_filter::MAP_SIZE as u32, 0);
 
 #[inline(always)]
-unsafe fn matches_filter(daddr: IpAddr) -> bool {
+unsafe fn matches_filter(_ctx: &XdpContext, daddr: IpAddr) -> bool {
     match daddr {
         IpAddr::V4(daddr) => {
-            if let Some(b) = FILTER_MAP.get(&daddr) {
-                *b == 1u8
-            } else {
-                false
+            for hash_offset in 0..bloom_filter::HASH_COUNT {
+                let hash = bloom_filter::hash(daddr, hash_offset);
+                let map_i = hash >> (bloom_filter::ADDRESS_BITS_CHUNK as u32);
+                let chunk_i = hash & (bloom_filter::ADDRESS_MASK_CHUNK as u32);
+                // info!(ctx, "{:ipv4} {} {} {}",daddr, hash, map_i, chunk_i);
+                let test = if let Some(b) = FILTER_MAP.get(map_i as u32) {
+                    let word_i = chunk_i & (bloom_filter::ADDRESS_MASK_WORD as u32);
+                    let chunk_i = (chunk_i as usize) >> bloom_filter::ADDRESS_BITS_WORD;
+                    // info!(ctx, "{} [{}]", word_i, b[chunk_i]);
+                    (b[chunk_i] >> (bloom_filter::ADDRESS_MASK_WORD as u32 - word_i)) & 1 == 1
+                } else {
+                    false
+                };
+                if !test {
+                    return false
+                }
             }
+            true
         }
         IpAddr::V6(_daddr) => {
             false // TODO
@@ -79,8 +95,7 @@ fn try_responder(ctx: XdpContext) -> Result<xdp_action::Type, xdp_action::Type>
         return Ok(xdp_action::XDP_PASS);
     }
 
-
-    if unsafe { !matches_filter(daddr) } {
+    if unsafe { !matches_filter(&ctx, daddr) } {
         return Ok(xdp_action::XDP_DROP);
     }
 
@@ -92,11 +107,11 @@ fn try_responder(ctx: XdpContext) -> Result<xdp_action::Type, xdp_action::Type>
     let tcp_syn = unsafe { (*tcp).syn() };
     let tcp_ack = unsafe { (*tcp).ack() };
 
-    match daddr {
-        IpAddr::V4(ip) => info!(&ctx, "Received packet with matching daddr: {:ipv4}", ip),
-        IpAddr::V6(ip) => unsafe { info!(&ctx, "Received packet with matching daddr: {:ipv6}", ip.in6_u.u6_addr8) }
-    }
-    info!(&ctx, "and tcp with syn: {}, ack: {}", tcp_syn, tcp_ack);
+    // match daddr {
+    //     IpAddr::V4(ip) => info!(&ctx, "Received packet with matching daddr: {:ipv4}", ip),
+    //     IpAddr::V6(ip) => unsafe { info!(&ctx, "Received packet with matching daddr: {:ipv6}", ip.in6_u.u6_addr8) }
+    // }
+    // info!(&ctx, "and tcp with syn: {}, ack: {}", tcp_syn, tcp_ack);
 
     if tcp_syn == 0 || tcp_ack != 0 {
         return Ok(xdp_action::XDP_PASS);

responder/Cargo.toml

@@ -21,8 +21,6 @@ clap = { version = "3.1", features = ["derive"] }
 env_logger = "0.9"
 log = "0.4"
 tokio = { version = "1.25", features = ["macros", "rt", "rt-multi-thread", "net", "signal"] }
-csv = "1.1"
-serde = { version = "1.0", features = ["derive"] }
 
 [[bin]]
 name = "responder"

responder/src/main.rs

@@ -1,6 +1,6 @@
-use std::{net::Ipv4Addr, path::PathBuf};
+use std::{path::PathBuf, time::Instant, fs::File, io::Read};
 
-use aya::{Bpf, maps::HashMap};
+use aya::{Bpf, maps::Array as BpfArray};
 use anyhow::{anyhow, Context};
 use aya::programs::{Xdp, XdpFlags};
 use aya_log::BpfLogger;
@@ -8,8 +8,7 @@ use clap::Parser;
 use env_logger::Env;
 use log::{info, warn};
 use tokio::signal;
-use csv::ReaderBuilder;
-use serde::Deserialize;
+use responder_common::bloom_filter;
 
 #[cfg(all(debug_assertions, not(feature = "default_artifact_build")))]
 const DEFAULT_TARGET: &str = "target/bpfel-unknown-none/debug";
@@ -27,18 +26,14 @@ struct Opt {
     #[clap(short, long, default_value = "syn")]
     scan_type: String,
     #[clap(short, long)]
-    csv: Option<String>,
+    bfb: Option<String>,
     #[clap(default_value = DEFAULT_TARGET, long)]
     target: PathBuf,
 }
 
-#[derive(Debug, Deserialize, Eq, PartialEq)]
-struct CsvRow {
-    saddr: Ipv4Addr,
-}
-
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
+    let start = Instant::now();
     let opt = Opt::parse();
 
     env_logger::Builder::from_env(Env::default().default_filter_or("info")).init();
@@ -68,23 +63,33 @@ This can happen if the loaded eBPF program has no log statements.", e);
 
     info!("Loaded {} XDP program", xdp_name);
 
-    let mut filter_map: HashMap<_, u32, u8> =
-        HashMap::try_from(bpf.map_mut("FILTER_MAP")
-                          .ok_or(anyhow!("Could not construct mutable FILTER_MAP"))?)?;
-
-    if let Some(csv_path) = opt.csv {
-        info!("Installing filter rules from {}", csv_path);
-        let mut reader = ReaderBuilder::new()
-            .has_headers(true)
-            .from_path(csv_path)?;
-        for record in reader.deserialize() {
-            let row: CsvRow = record?;
-            filter_map.insert(u32::from(row.saddr), 1u8, 0)?;
-            // TODO replace with BPF_MAP_UPDATE_BATCH
+
+
+    if let Some(bfb_path) = opt.bfb {
+        info!("Installing filter rules from {}", bfb_path);
+        let mut filter_map: BpfArray<_, [u8; bloom_filter::CHUNK_BYTES]> =
+            BpfArray::try_from(bpf.map_mut("FILTER_MAP")
+                               .ok_or(anyhow!("Could not construct mutable FILTER_MAP"))?)?;
+        let mut bloom_filter_binary = File::open(bfb_path)?;
+        let mut chunk = [0 as u8; bloom_filter::CHUNK_BYTES];
+
+        let mut i = 0;
+        loop {
+            let read = bloom_filter_binary.read(&mut chunk)?;
+            if read > 0 {
+                filter_map.set(i, chunk, 0)?;
+            }
+
+            if read < bloom_filter::CHUNK_BYTES {
+                break;
+            }
+            i += 1;
+
         }
     };
 
-    info!("press ctrl-c to exit");
+    info!("Loading XDP program took {:?}", start.elapsed());
+    println!("press ctrl-c to exit");
     signal::ctrl_c().await?;
     info!("Exiting...");
 

xtask/Cargo.toml

@@ -6,5 +6,7 @@ edition = "2021"
 [dependencies]
 anyhow = "1"
 clap = { version = "3.1", features = ["derive"] }
-# aya-tool = { path = "/home/niels/files/temp/aya/aya-tool" }
 aya-tool = { git = "https://github.com/aya-rs/aya" }
+responder-common = { path = "../responder-common", features=["user"] }
+env_logger = "0.9"
+log = "0.4"

xtask/src/filter.rs

@@ -0,0 +1,72 @@
+use std::{path::PathBuf, time::Instant, net::Ipv4Addr, fs::File, io::{Write, BufReader, BufRead}};
+
+use anyhow::anyhow;
+use clap::Parser;
+use log::info;
+use responder_common::bloom_filter;
+
+#[derive(Debug, Parser)]
+pub struct Options {
+    pub ip_path: PathBuf,
+    pub out_path: Option<PathBuf>,
+    #[clap(short, long)]
+    pub force: bool,
+}
+
+pub fn build_filter(opts: Options) -> Result<(), anyhow::Error> {
+    let out_path = opts.out_path.unwrap_or_else(|| {
+        let mut ext = opts.ip_path.extension().unwrap_or_default().to_os_string();
+        ext.push(".bfb");
+        opts.ip_path.with_extension(ext)
+    });
+
+    if out_path.exists() {
+        if !opts.force {
+            return Err(anyhow!("Output path {} already exists, use --force/-f to overwrite", out_path.display()));
+        } else if !out_path.is_file() {
+            return Err(anyhow!("Can not force overwrite output path {}: is not a file", out_path.display()));
+        }
+    }
+
+    let start = Instant::now();
+
+    // let mut addresses = Vec::new();
+    let mut bloom_filter: Vec<u8> = Vec::with_capacity(bloom_filter::BITS>>3);
+    bloom_filter.resize(bloom_filter::BITS >> 3, 0);
+
+    let f = File::open(opts.ip_path)?;
+    let reader = BufReader::new(f);
+    let mut clash_count = 0;
+    for (i, line) in reader.lines().enumerate() {
+        if i % 1_000_000 == 0{
+            info!("processed {}M ips; {} clashes",i / 1_000_000, clash_count);
+        }
+        let line = line?;
+        let key = u32::from(line.parse::<Ipv4Addr>()?);
+        // addresses.push(key);
+        let mut clashes = true;
+        for i in 0..bloom_filter::HASH_COUNT {
+            let hash = bloom_filter::hash(key, i);
+            let i = hash >> 3;
+            let byte_i = hash & 7;
+            // info!("{} {} {} {}", line, hash, i, byte_i);
+            clashes = clashes && ((bloom_filter[i as usize] >> (7 - byte_i)) & 1 == 1);
+            bloom_filter[i as usize] |= 0x80 >> byte_i;
+        }
+        if clashes {
+            clash_count += 1;
+        }
+    }
+    info!("clash count: {}", clash_count);
+
+    let filter = start.elapsed();
+
+    let mut file = File::create(out_path)?;
+    let byte_slice: &[u8] = bloom_filter.as_slice();
+    file.write_all(byte_slice)?;
+
+    let end = start.elapsed();
+
+    info!("build filter: {:?} write: {:?} | total: {:?}", filter, end-filter, end);
+    Ok(())
+}

xtask/src/main.rs

@@ -2,10 +2,12 @@ mod build_ebpf;
 mod build;
 mod run;
 mod codegen;
+mod filter;
 
 use std::process::exit;
 
 use clap::Parser;
+use env_logger::Env;
 
 
 #[derive(Debug, Parser)]
@@ -18,11 +20,14 @@ pub struct Options {
 enum Command {
     BuildEbpf(build_ebpf::Options),
     BuildArtifacts(build::Options),
+    BuildFilterFile(filter::Options),
     Run(run::Options),
     CodeGen,
 }
 
 fn main() {
+    env_logger::Builder::from_env(Env::default().default_filter_or("info")).init();
+
     let opts = Options::parse();
 
     use Command::*;
@@ -31,6 +36,7 @@ fn main() {
         BuildArtifacts(opts) => build::build(opts),
         Run(opts) => run::run(opts),
         CodeGen => codegen::generate(),
+        BuildFilterFile(opts) => filter::build_filter(opts),
     };
 
     if let Err(e) = ret {