rustflow_export 0.20.0

// Manual AF_PACKET implementation with PACKET_MMAP (TPACKET_V3).
// We don't use the `af_packet` crate because it has a musl compilation bug:
// ioctl request type mismatch (u64 vs i32) that prevents cross-compilation.
//
// TPACKET_V3 is block-oriented: the kernel packs frames into a block and
// hands over the whole block, so reading costs a wakeup per block instead
// of one per frame.

use std::ffi::CString;
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
use std::sync::atomic::{AtomicU32, Ordering};
use std::{io, mem, ptr, slice};

use anyhow::{Result, anyhow};
use log::{info, warn};

use super::{Capture, Frame, Link, capture_time};

const TPACKET_V3: libc::c_int = 2;

const BLOCK_SIZE: u32 = 1 << 16;
const BLOCK_NR: u32 = 64;
const FRAME_SIZE: u32 = 2048;
const FRAME_NR: u32 = (BLOCK_SIZE / FRAME_SIZE) * BLOCK_NR;

/// A block that is not full is handed over after this long, so packets on
/// a quiet link are not held back.
const BLOCK_TIMEOUT_MS: u32 = 100;

/// Bytes of each frame copied into the ring: the headers are all that is
/// parsed, and a V3 frame is otherwise as large as the packet.
pub const SNAPLEN: u32 = 2048;

/// Classic BPF `ret k`: accept the packet, truncated to `k` bytes.
const BPF_RET: u16 = 0x06;

pub struct AfPacket {
    socket: OwnedFd,
    ring: Ring,
    block_idx: u32,
    /// Whether a block is held. It is handed back on the read after its
    /// last frame, because that frame is still borrowed by the caller.
    block_held: bool,
    /// Frames still to read in the block held.
    frames_left: u32,
    /// Ring offset of the next frame in that block.
    frame_offset: usize,
}

struct Ring {
    ptr: *mut u8,
    len: usize,
}

impl Ring {
    fn map(socket: &OwnedFd, len: usize) -> io::Result<Self> {
        // SAFETY: a new shared mapping of the socket's ring, at an address the
        // kernel picks; no memory of this process is touched.
        let ptr = unsafe {
            libc::mmap(
                ptr::null_mut(),
                len,
                libc::PROT_READ | libc::PROT_WRITE,
                libc::MAP_SHARED,
                socket.as_raw_fd(),
                0,
            )
        };
        if ptr == libc::MAP_FAILED {
            return Err(io::Error::last_os_error());
        }
        Ok(Self {
            ptr: ptr as *mut u8,
            len,
        })
    }
}

impl Drop for Ring {
    fn drop(&mut self) {
        // SAFETY: `ptr` and `len` are the mapping `map` made. Frames borrow
        // the capture that owns the ring, so none outlives it.
        unsafe {
            libc::munmap(self.ptr as *mut libc::c_void, self.len);
        }
    }
}

fn set_option<T>(
    socket: &OwnedFd,
    level: libc::c_int,
    name: libc::c_int,
    value: &T,
) -> io::Result<()> {
    // SAFETY: the kernel reads `size_of::<T>()` bytes from `value`, a live `T`.
    let ret = unsafe {
        libc::setsockopt(
            socket.as_raw_fd(),
            level,
            name,
            ptr::from_ref(value).cast(),
            mem::size_of::<T>() as libc::socklen_t,
        )
    };
    if ret < 0 {
        return Err(io::Error::last_os_error());
    }
    Ok(())
}

impl AfPacket {
    /// `filter` is a classic BPF program to run in the kernel, whose
    /// accepting returns already carry the snap length; without one every
    /// frame is taken, cut at the snap length.
    pub fn new(
        interface: &str,
        promiscuous: bool,
        filter: Option<&[libc::sock_filter]>,
    ) -> Result<Self> {
        info!("Opening AF_PACKET capture on interface: {}", interface);

        let protocol = (libc::ETH_P_ALL as u16).to_be();
        // SAFETY: takes no pointers.
        let fd = unsafe { libc::socket(libc::AF_PACKET, libc::SOCK_RAW, protocol as libc::c_int) };
        if fd < 0 {
            return Err(anyhow!(
                "Failed to create socket: {}",
                io::Error::last_os_error()
            ));
        }
        // SAFETY: `fd` was just opened, and nothing else owns it.
        let socket = unsafe { OwnedFd::from_raw_fd(fd) };

        set_option(&socket, libc::SOL_PACKET, libc::PACKET_VERSION, &TPACKET_V3)
            .map_err(|e| anyhow!("Failed to set TPACKET_V3: {}", e))?;

        let snap = [libc::sock_filter {
            code: BPF_RET,
            jt: 0,
            jf: 0,
            k: SNAPLEN,
        }];
        let program = filter.unwrap_or(&snap);
        let filter = libc::sock_fprog {
            len: u16::try_from(program.len())
                .map_err(|_| anyhow!("The packet filter has too many instructions"))?,
            filter: program.as_ptr().cast_mut(),
        };
        set_option(&socket, libc::SOL_SOCKET, libc::SO_ATTACH_FILTER, &filter)
            .map_err(|e| anyhow!("Failed to attach the packet filter: {}", e))?;

        let req = libc::tpacket_req3 {
            tp_block_size: BLOCK_SIZE,
            tp_block_nr: BLOCK_NR,
            tp_frame_size: FRAME_SIZE,
            tp_frame_nr: FRAME_NR,
            tp_retire_blk_tov: BLOCK_TIMEOUT_MS,
            tp_sizeof_priv: 0,
            tp_feature_req_word: 0,
        };
        set_option(&socket, libc::SOL_PACKET, libc::PACKET_RX_RING, &req)
            .map_err(|e| anyhow!("Failed to setup ring buffer: {}", e))?;

        let ring = Ring::map(&socket, (BLOCK_SIZE * BLOCK_NR) as usize)
            .map_err(|e| anyhow!("Failed to mmap ring buffer: {}", e))?;

        let ifindex = get_interface_index(&socket, interface)?;

        let addr = libc::sockaddr_ll {
            sll_family: libc::AF_PACKET as u16,
            sll_protocol: protocol,
            sll_ifindex: ifindex,
            sll_hatype: 0,
            sll_pkttype: 0,
            sll_halen: 0,
            sll_addr: [0; 8],
        };
        // SAFETY: the kernel reads one `sockaddr_ll`, `addr`.
        let ret = unsafe {
            libc::bind(
                socket.as_raw_fd(),
                ptr::from_ref(&addr).cast(),
                mem::size_of::<libc::sockaddr_ll>() as libc::socklen_t,
            )
        };
        if ret < 0 {
            return Err(anyhow!(
                "Failed to bind to interface: {}",
                io::Error::last_os_error()
            ));
        }

        if promiscuous {
            let mreq = libc::packet_mreq {
                mr_ifindex: ifindex,
                mr_type: libc::PACKET_MR_PROMISC as libc::c_ushort,
                mr_alen: 0,
                mr_address: [0; 8],
            };
            set_option(
                &socket,
                libc::SOL_PACKET,
                libc::PACKET_ADD_MEMBERSHIP,
                &mreq,
            )
            .map_err(|e| {
                anyhow!(
                    "Failed to enable promiscuous mode on '{}': {}",
                    interface,
                    e
                )
            })?;

            info!("Promiscuous mode enabled on {}", interface);
        }

        info!(
            "AF_PACKET ring buffer ready: {} blocks of {} KiB",
            BLOCK_NR,
            BLOCK_SIZE / 1024
        );

        Ok(Self {
            socket,
            ring,
            block_idx: 0,
            block_held: false,
            frames_left: 0,
            frame_offset: 0,
        })
    }

    fn block_offset(&self) -> usize {
        (self.block_idx * BLOCK_SIZE) as usize
    }

    fn block(&self) -> *mut libc::tpacket_hdr_v1 {
        // SAFETY: `block_idx < BLOCK_NR`, so the offset lies inside the ring,
        // where each block starts with its descriptor. No reference is made.
        unsafe {
            let desc = self.ring.ptr.add(self.block_offset()) as *mut libc::tpacket_block_desc;
            ptr::addr_of_mut!((*desc).hdr.bh1)
        }
    }

    /// The kernel and this thread pass the block back and forth through
    /// its status word.
    fn block_status(&self) -> &AtomicU32 {
        // SAFETY: the status word is aligned, lives as long as the ring, which
        // `self` owns, and is only accessed atomically, here and by the kernel.
        unsafe { AtomicU32::from_ptr(ptr::addr_of_mut!((*self.block()).block_status)) }
    }

    fn block_ready(&self) -> bool {
        self.block_status().load(Ordering::Acquire) & libc::TP_STATUS_USER != 0
    }

    /// Takes the next block once the kernel has handed it over, waiting up
    /// to a second for it.
    fn acquire_block(&mut self) -> bool {
        if !self.block_ready() {
            let mut pfd = libc::pollfd {
                fd: self.socket.as_raw_fd(),
                events: libc::POLLIN,
                revents: 0,
            };
            // SAFETY: `pfd` is the one `pollfd` the count says.
            let ret = unsafe { libc::poll(&mut pfd, 1, 1000) };
            if ret > 0 && pfd.revents & libc::POLLERR != 0 {
                // A pending socket error, such as ENETDOWN once the
                // interface goes down, makes poll return at once until it is
                // read.
                match take_error(&self.socket) {
                    Ok(Some(e)) | Err(e) => warn!("Capture error: {}", e),
                    Ok(None) => {}
                }
                return false;
            }
            if ret <= 0 || !self.block_ready() {
                return false;
            }
        }

        // SAFETY: the block is ours: its status was seen with TP_STATUS_USER,
        // with acquire ordering, so the kernel is done writing it.
        let (frames, first_frame) = unsafe {
            let block = self.block();
            ((*block).num_pkts, (*block).offset_to_first_pkt)
        };
        self.block_held = true;
        if frames == 0 {
            // Handed over empty by the block timeout.
            self.release_block();
            return false;
        }

        self.frames_left = frames;
        self.frame_offset = self.block_offset() + first_frame as usize;
        true
    }

    fn release_block(&mut self) {
        self.block_status()
            .store(libc::TP_STATUS_KERNEL, Ordering::Release);
        self.block_idx = (self.block_idx + 1) % BLOCK_NR;
        self.block_held = false;
    }
}

impl Capture for AfPacket {
    fn link(&self) -> Link {
        Link::Ethernet
    }

    fn next_frame(&mut self) -> Option<Frame<'_>> {
        loop {
            if self.frames_left == 0 {
                if self.block_held {
                    self.release_block();
                }
                if !self.acquire_block() {
                    return None;
                }
            }

            // SAFETY: the kernel's offsets keep `frame_offset` on a frame of
            // the block held while frames are left.
            let frame = unsafe { self.ring.ptr.add(self.frame_offset) };
            // SAFETY: a frame starts with its header, aligned, and the kernel
            // does not write a block it has handed over.
            let hdr = unsafe { &*(frame as *const libc::tpacket3_hdr) };
            self.frames_left -= 1;
            self.frame_offset += hdr.tp_next_offset as usize;

            // SAFETY: `frame` is a frame of the block held.
            if unsafe { is_loopback_echo(frame) } {
                continue;
            }
            // SAFETY: the kernel put `tp_snaplen` bytes of the packet at
            // `tp_mac`, within the frame. The slice borrows `self`, so the
            // block is not handed back while it lives.
            let data = unsafe {
                slice::from_raw_parts(frame.add(hdr.tp_mac as usize), hdr.tp_snaplen as usize)
            };
            return Some(Frame {
                data,
                length: hdr.tp_len,
                time: capture_time(i64::from(hdr.tp_sec), hdr.tp_nsec),
            });
        }
    }
}

/// A loopback device hands every packet to a packet socket twice, as sent
/// and as received; like libpcap, keep only the received copy.
///
/// # Safety
///
/// `frame` must point to a frame of a TPACKET_V3 block the kernel has
/// handed over.
unsafe fn is_loopback_echo(frame: *const u8) -> bool {
    let offset = libc::TPACKET3_HDRLEN - mem::size_of::<libc::sockaddr_ll>();
    // SAFETY: the kernel puts the frame's link-layer address right after its
    // header, aligned, ending at `TPACKET3_HDRLEN`.
    let sll = unsafe { &*(frame.add(offset) as *const libc::sockaddr_ll) };
    sll.sll_pkttype == libc::PACKET_OUTGOING && sll.sll_hatype == libc::ARPHRD_LOOPBACK
}

/// Reads and clears the socket's pending error.
fn take_error(socket: &OwnedFd) -> io::Result<Option<io::Error>> {
    let mut error: libc::c_int = 0;
    let mut length = mem::size_of::<libc::c_int>() as libc::socklen_t;
    // SAFETY: the kernel writes at most `length` bytes, the size of `error`.
    let ret = unsafe {
        libc::getsockopt(
            socket.as_raw_fd(),
            libc::SOL_SOCKET,
            libc::SO_ERROR,
            ptr::from_mut(&mut error).cast(),
            &mut length,
        )
    };
    if ret < 0 {
        return Err(io::Error::last_os_error());
    }
    Ok((error != 0).then(|| io::Error::from_raw_os_error(error)))
}

fn get_interface_index(socket: &OwnedFd, interface: &str) -> Result<libc::c_int> {
    let ifname = CString::new(interface)?;
    // SAFETY: `ifreq` is plain data, for which all zeros is a valid value.
    let mut ifr: libc::ifreq = unsafe { mem::zeroed() };

    // Copy interface name (max 15 chars + null)
    let name_bytes = ifname.as_bytes_with_nul();
    let copy_len = name_bytes.len().min(libc::IFNAMSIZ);
    // SAFETY: `copy_len` is within both the name and `ifr_name`, which is
    // IFNAMSIZ long.
    unsafe {
        ptr::copy_nonoverlapping(
            name_bytes.as_ptr(),
            ifr.ifr_name.as_mut_ptr().cast(),
            copy_len,
        );
    }

    // ioctl request type differs between glibc (c_ulong) and musl (c_int)
    #[cfg(target_env = "musl")]
    let request = libc::SIOCGIFINDEX as libc::c_int;
    #[cfg(not(target_env = "musl"))]
    let request = libc::SIOCGIFINDEX as libc::c_ulong;

    // SAFETY: SIOCGIFINDEX reads and writes one `ifreq`, `ifr`.
    let ret = unsafe { libc::ioctl(socket.as_raw_fd(), request, &mut ifr) };
    if ret < 0 {
        return Err(anyhow!(
            "Interface '{}' not found: {}",
            interface,
            io::Error::last_os_error()
        ));
    }

    // SAFETY: a successful SIOCGIFINDEX sets `ifru_ifindex`.
    Ok(unsafe { ifr.ifr_ifru.ifru_ifindex })
}