feat: initial commit

the initial basic implementation is here:

1. capture all signals through the generate_signal trace.
2. pipes the messages from the threads to the main task.
3. uses sync channels.

Author: Ricardo Maraschini

.gitignore

@@ -0,0 +1,16 @@
+### https://raw.github.com/github/gitignore/master/Rust.gitignore
+
+# Generated by Cargo
+# will have compiled files and executables
+debug/
+target/
+
+# Remove Cargo.lock from gitignore if creating an executable, leave it for libraries
+# More information here https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html
+Cargo.lock
+
+# These are backup files generated by rustfmt
+**/*.rs.bk
+
+.vim/
+.vscode/

Cargo.toml

@@ -0,0 +1,2 @@
+[workspace]
+members = ["signals", "signals-common", "xtask"]

README.md

@@ -1,2 +1,32 @@
-# ebpf-signals
-An eBPF engine for capturing and processing signals
+# Signals
+
+An eBPF based engine to capture and process signals being sent.
+
+
+## Prerequisites
+
+1. Install a rust stable toolchain: `rustup install stable`
+2. Install a rust nightly toolchain: `rustup install nightly`
+3. Install bpf-linker: `cargo install bpf-linker`
+4. Install rust-src: `rustup component add rust-src --toolchain nightly-x86_64-unknown-linux-gnu`
+
+## Build eBPF
+
+```bash
+cargo xtask build-ebpf
+```
+
+To perform a release build you can use the `--release` flag.
+You may also change the target architecture with the `--target` flag
+
+## Build Userspace
+
+```bash
+cargo build
+```
+
+## Run
+
+```bash
+sudo RUST_LOG=debug ./target/debug/signals
+```

signals-common/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "signals-common"
+version = "0.1.0"
+edition = "2021"
+
+[features]
+default = []
+user = [ "aya" ]
+
+[dependencies]
+aya = { version = ">=0.11", optional=true }
+
+[lib]
+path = "src/lib.rs"

signals-common/src/lib.rs

@@ -0,0 +1,8 @@
+#![no_std]
+
+#[repr(C)]
+#[derive(Clone, Copy, Debug)]
+pub struct Signal {
+    pub signr: i32,
+    pub pid: u32,
+}

signals-ebpf/.cargo/config.toml

@@ -0,0 +1,6 @@
+[build]
+target-dir = "../target"
+target = "bpfel-unknown-none"
+
+[unstable]
+build-std = ["core"]

signals-ebpf/Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+name = "signals-ebpf"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+aya-bpf = { git = "https://github.com/aya-rs/aya", branch = "main" }
+aya-log-ebpf = { git = "https://github.com/aya-rs/aya", branch = "main" }
+signals-common = { path = "../signals-common" }
+
+[[bin]]
+name = "signals"
+path = "src/main.rs"
+
+[profile.dev]
+opt-level = 3
+debug = false
+debug-assertions = false
+overflow-checks = false
+lto = true
+panic = "abort"
+incremental = false
+codegen-units = 1
+rpath = false
+
+[profile.release]
+lto = true
+panic = "abort"
+codegen-units = 1
+
+[workspace]
+members = []

signals-ebpf/rust-toolchain.toml

@@ -0,0 +1,2 @@
+[toolchain]
+channel="nightly"

signals-ebpf/src/main.rs

@@ -0,0 +1,53 @@
+#![no_std]
+#![no_main]
+
+use aya_bpf::macros::map;
+use aya_bpf::macros::tracepoint;
+use aya_bpf::maps::PerfEventArray;
+use aya_bpf::programs::TracePointContext;
+use aya_log_ebpf::debug;
+use signals_common::Signal;
+
+#[map(name = "SIGNALS")] //
+static mut SIGNALS: PerfEventArray<Signal> = PerfEventArray::<Signal>::with_max_entries(1024, 0);
+
+// these offsets are obtained by reading the following file:
+// /sys/kernel/debug/tracing/events/signal/signal_generate/format
+const SIGNAL_OFFSET: usize = 8;
+const PID_OFFSET: usize = 36;
+
+#[tracepoint(name = "signals")]
+pub fn signals(ctx: TracePointContext) -> u32 {
+    match try_signals(ctx) {
+        Ok(ret) => ret,
+        Err(ret) => ret,
+    }
+}
+
+fn try_signals(ctx: TracePointContext) -> Result<u32, u32> {
+    let signr: i32 = unsafe {
+        match ctx.read_at(SIGNAL_OFFSET) {
+            Ok(s) => s,
+            Err(errn) => return Err(errn as u32),
+        }
+    };
+
+    let pid: u32 = unsafe {
+        match ctx.read_at(PID_OFFSET) {
+            Ok(s) => s,
+            Err(errn) => return Err(errn as u32),
+        }
+    };
+
+    let s = Signal { signr, pid };
+    unsafe {
+        debug!(&ctx, "ebpf: enqueued signal {} for {}", signr, pid);
+        SIGNALS.output(&ctx, &s, 0);
+    }
+    Ok(0)
+}
+
+#[panic_handler]
+fn panic(_info: &core::panic::PanicInfo) -> ! {
+    unsafe { core::hint::unreachable_unchecked() }
+}

signals/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "signals"
+version = "0.1.0"
+edition = "2021"
+publish = false
+
+[dependencies]
+aya = { version = ">=0.11", features=["async_tokio"] }
+aya-log = "0.1"
+signals-common = { path = "../signals-common", features=["user"] }
+anyhow = "1.0.42"
+clap = { version = "3.1", features = ["derive"] }
+env_logger = "0.9"
+log = "0.4"
+tokio = { version = "1.18", features = ["full"] }
+bytes = "1"
+prost = "0.10.4"
+tonic = "0.7.2"
+
+[[bin]]
+name = "signals"
+path = "src/main.rs"
+
+[build-dependencies]
+tonic-build = "0.7.2"

signals/src/emitters.rs

@@ -0,0 +1,93 @@
+use aya::include_bytes_aligned;
+use aya::maps::perf::AsyncPerfEventArray;
+use aya::programs::TracePoint;
+use aya::util::online_cpus;
+use aya::Bpf;
+use aya_log::BpfLogger;
+use bytes::BytesMut;
+use log::debug;
+use log::error;
+use signals_common::Signal;
+use std::mem::size_of;
+use std::sync::mpsc::SyncSender;
+use tokio::task;
+
+pub struct SignalEmitter {
+    bpf: Bpf,
+    dst: SyncSender<Signal>,
+}
+
+impl SignalEmitter {
+    // new returns a new Signal that loads a bpf program from disk and loads it
+    // into the kernel. signals are written to the destination channel.
+    pub fn new(dst: SyncSender<Signal>) -> Result<Self, anyhow::Error> {
+        #[cfg(debug_assertions)]
+        let mut bpf = Bpf::load(include_bytes_aligned!(
+            "../../target/bpfel-unknown-none/debug/signals"
+        ))?;
+
+        #[cfg(not(debug_assertions))]
+        let mut bpf = Bpf::load(include_bytes_aligned!(
+            "../../target/bpfel-unknown-none/release/signals"
+        ))?;
+
+        BpfLogger::init(&mut bpf).unwrap();
+
+        let program: &mut TracePoint = bpf.program_mut("signals").unwrap().try_into()?;
+        program.load()?;
+        program.attach("signal", "signal_generate")?;
+
+        Ok(Self { bpf, dst })
+    }
+
+    // attach starts to read signal events from the kernel and pipe them through the
+    // destination channel. spawns a task per cpu, each task process events from its
+    // own perf array.
+    pub fn attach(&mut self) -> Result<(), anyhow::Error> {
+        let cpus = online_cpus()?;
+        let mut senders: Vec<SyncSender<Signal>> = vec![];
+        for _ in 0..cpus.len() {
+            senders.push(self.dst.clone());
+        }
+
+        let signal_struct_size: usize = size_of::<Signal>();
+        let mut perf_array = AsyncPerfEventArray::try_from(self.bpf.map_mut("SIGNALS")?)?;
+        for cpu_id in cpus {
+            debug!("spawning task for cpu {}", cpu_id);
+            let mut parray = perf_array.open(cpu_id, None)?;
+            let txcopy = senders.pop().unwrap();
+            task::spawn(async move {
+                debug!("task for cpu awaiting for events {}", cpu_id);
+                let mut buffers = (0..100)
+                    .map(|_| BytesMut::with_capacity(signal_struct_size))
+                    .collect::<Vec<_>>();
+
+                loop {
+                    let events = match parray.read_events(&mut buffers).await {
+                        Ok(events) => events,
+                        Err(error) => {
+                            error!("fail to read events from the perf, bailing out: {}", error);
+                            return;
+                        }
+                    };
+
+                    if events.lost > 0 {
+                        error!("queues are getting full, lost {} signals", events.lost);
+                    }
+
+                    for i in 0..events.read {
+                        let buf = &mut buffers[i];
+                        let ptr = buf.as_ptr() as *const Signal;
+                        let signal = unsafe { ptr.read_unaligned() };
+                        match txcopy.send(signal) {
+                            Ok(_) => continue,
+                            Err(err) => error!("failed to send signal internally: {}", err),
+                        }
+                    }
+                }
+            });
+        }
+
+        Ok(())
+    }
+}

signals/src/main.rs

@@ -0,0 +1,20 @@
+pub mod emitters;
+
+use log::info;
+use std::sync::mpsc::sync_channel;
+
+#[tokio::main]
+async fn main() -> Result<(), anyhow::Error> {
+    env_logger::init();
+
+    let (tx, rx) = sync_channel(100);
+
+    let mut _bpf = emitters::SignalEmitter::new(tx)?;
+    _bpf.attach()?;
+
+    info!("tasks started, awaiting for events");
+    for msg in rx {
+        info!("{:?}", msg);
+    }
+    Ok(())
+}

xtask/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "xtask"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+anyhow = "1"
+clap = { version = "3.1", features = ["derive"] }

xtask/src/build_ebpf.rs

@@ -0,0 +1,64 @@
+use std::path::PathBuf;
+use std::process::Command;
+
+use clap::Parser;
+
+#[derive(Debug, Copy, Clone)]
+pub enum Architecture {
+    BpfEl,
+    BpfEb,
+}
+
+impl std::str::FromStr for Architecture {
+    type Err = String;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Ok(match s {
+            "bpfel-unknown-none" => Architecture::BpfEl,
+            "bpfeb-unknown-none" => Architecture::BpfEb,
+            _ => return Err("invalid target".to_owned()),
+        })
+    }
+}
+
+impl std::fmt::Display for Architecture {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(match self {
+            Architecture::BpfEl => "bpfel-unknown-none",
+            Architecture::BpfEb => "bpfeb-unknown-none",
+        })
+    }
+}
+
+#[derive(Debug, Parser)]
+pub struct Options {
+    /// Set the endianness of the BPF target
+    #[clap(default_value = "bpfel-unknown-none", long)]
+    pub target: Architecture,
+    /// Build the release target
+    #[clap(long)]
+    pub release: bool,
+}
+
+pub fn build_ebpf(opts: Options) -> Result<(), anyhow::Error> {
+    let dir = PathBuf::from("signals-ebpf");
+    let target = format!("--target={}", opts.target);
+    let mut args = vec![
+        "+nightly",
+        "build",
+        "--verbose",
+        target.as_str(),
+        "-Z",
+        "build-std=core",
+    ];
+    if opts.release {
+        args.push("--release")
+    }
+    let status = Command::new("cargo")
+        .current_dir(&dir)
+        .args(&args)
+        .status()
+        .expect("failed to build bpf program");
+    assert!(status.success());
+    Ok(())
+}