ferranet 0.2.0

//! An in-memory, privilege-free datalink channel for tests.
//!
//! [`channel`] returns a real [`Sender`]/[`Receiver`] pair (the same types the `AF_PACKET` backend
//! produces) wired to FIFO queues instead of a NIC, plus two handles:
//!
//! * [`Injector`] — push frames the [`Receiver`] will deliver (or inject a simulated I/O error),
//! * [`Drain`] — read frames the [`Sender`] transmitted.
//!
//! Receive readiness is backed by a pipe, so the dummy works with the blocking API, `poll`, and —
//! with the `tokio` feature — the async API, exactly like a real channel. This lets downstream
//! code test packet handling deterministically without `CAP_NET_RAW`, a `veth` pair, or a network
//! namespace.
//!
//! ```
//! # #[cfg(target_os = "linux")]
//! # fn main() -> ferranet::Result<()> {
//! let mut net = ferranet::dummy::channel()?;
//! net.inject.inject(b"\xff\xff\xff\xff\xff\xff... a frame".to_vec());
//! let block = net.rx.recv_block()?;
//! assert_eq!(block.len(), 1);
//!
//! net.tx.send(b"outgoing frame")?;
//! assert_eq!(net.drain.try_recv().unwrap(), b"outgoing frame");
//! # Ok(())
//! # }
//! # #[cfg(not(target_os = "linux"))]
//! # fn main() {}
//! ```

use std::collections::VecDeque;
use std::io;
use std::os::fd::{AsRawFd, BorrowedFd, FromRawFd, OwnedFd, RawFd};
use std::sync::{Arc, Mutex};

use crate::channel::{Receiver, Sender};
use crate::error::Result;
use crate::interface::{IfIndex, Interface, MacAddr};

/// Shared queue of frames waiting to be received (or injected errors).
pub(crate) type RxQueue = Arc<Mutex<VecDeque<io::Result<Vec<u8>>>>>;
/// Shared queue of frames that were transmitted.
pub(crate) type SentQueue = Arc<Mutex<VecDeque<Vec<u8>>>>;

/// A fake datalink channel and its test handles.
#[derive(Debug)]
pub struct DummyChannel {
    /// Transmit half — frames sent here land in [`DummyChannel::drain`].
    pub tx: Sender,
    /// Receive half — yields frames pushed via [`DummyChannel::inject`].
    pub rx: Receiver,
    /// Pushes frames for `rx` to receive.
    pub inject: Injector,
    /// Reads frames that `tx` sent.
    pub drain: Drain,
}

/// Creates an in-memory datalink channel backed by FIFO queues.
pub fn channel() -> Result<DummyChannel> {
    let (read, write) = make_pipe()?;
    let rx_queue: RxQueue = Arc::new(Mutex::new(VecDeque::new()));
    let sent: SentQueue = Arc::new(Mutex::new(VecDeque::new()));

    // An inert pipe end gives the sender a real fd for `as_fd`; it is never written to.
    let (_sender_read, sender_fd) = make_pipe()?;

    let tx = Sender::from_dummy(sent.clone(), sender_fd);
    let rx = Receiver::from_dummy(rx_queue.clone(), read);

    Ok(DummyChannel {
        tx,
        rx,
        inject: Injector { queue: rx_queue, notify: write },
        drain: Drain { sent },
    })
}

/// A fake asynchronous datalink channel and its test handles (feature `tokio`).
///
/// The async counterpart of [`DummyChannel`]: the same queues and readiness pipe, surfaced
/// through the real [`AsyncSender`]/[`AsyncReceiver`] types.
#[cfg(feature = "tokio")]
#[derive(Debug)]
pub struct AsyncDummyChannel {
    /// Transmit half — frames sent here land in [`AsyncDummyChannel::drain`].
    pub tx: crate::async_channel::AsyncSender,
    /// Receive half — yields frames pushed via [`AsyncDummyChannel::inject`].
    pub rx: crate::async_channel::AsyncReceiver,
    /// Pushes frames for `rx` to receive.
    pub inject: Injector,
    /// Reads frames that `tx` sent.
    pub drain: Drain,
}

/// Creates an in-memory asynchronous datalink channel backed by FIFO queues (feature `tokio`).
///
/// Must be called from within a Tokio runtime context (the readiness descriptors are registered
/// with the reactor on construction), like
/// [`ChannelBuilder::build_async`](crate::ChannelBuilder::build_async).
#[cfg(feature = "tokio")]
pub fn channel_async() -> Result<AsyncDummyChannel> {
    use crate::async_channel::{AsyncReceiver, AsyncSender};
    use crate::channel::{RxBackend, TxBackend};

    let (read, write) = make_pipe()?;
    let rx_queue: RxQueue = Arc::new(Mutex::new(VecDeque::new()));
    let sent: SentQueue = Arc::new(Mutex::new(VecDeque::new()));

    // An inert pipe end gives the sender a real fd for readiness registration; the dummy TX
    // queue is unbounded, so writability is never actually awaited.
    let (_sender_read, sender_fd) = make_pipe()?;

    let tx = AsyncSender::new(TxBackend::Dummy { sent: sent.clone(), fd: sender_fd })?;
    let rx = AsyncReceiver::new(RxBackend::Dummy {
        queue: rx_queue.clone(),
        readiness: read,
        last: Vec::new(),
    })?;

    Ok(AsyncDummyChannel {
        tx,
        rx,
        inject: Injector { queue: rx_queue, notify: write },
        drain: Drain { sent },
    })
}

/// Pushes frames into the fake network for the [`Receiver`] to deliver.
///
/// Dropping the injector closes the network: a `Receiver` that has drained every queued frame then
/// observes end-of-stream (`recv_block` returns [`io::ErrorKind::UnexpectedEof`]).
#[derive(Debug)]
pub struct Injector {
    queue: RxQueue,
    notify: OwnedFd,
}

impl Injector {
    /// Queues a frame to be received.
    pub fn inject(&self, frame: Vec<u8>) {
        self.push(Ok(frame));
    }

    /// Queues a simulated receive error to be returned by the next `recv_block`.
    pub fn inject_error(&self, err: io::Error) {
        self.push(Err(err));
    }

    fn push(&self, item: io::Result<Vec<u8>>) {
        self.queue.lock().expect("dummy queue poisoned").push_back(item);
        // Wake any receiver blocked on readiness. Best-effort: a full pipe just means the receiver
        // hasn't drained yet, and it re-checks the queue regardless.
        let byte = [1u8];
        // SAFETY: writing one byte to a valid, non-blocking pipe fd we own.
        unsafe {
            libc::write(self.notify.as_raw_fd(), byte.as_ptr().cast(), 1);
        }
    }
}

/// Reads frames that the [`Sender`] transmitted into the fake network.
#[derive(Debug)]
pub struct Drain {
    sent: SentQueue,
}

impl Drain {
    /// Removes and returns the oldest transmitted frame, if any.
    #[must_use]
    pub fn try_recv(&self) -> Option<Vec<u8>> {
        self.sent.lock().expect("dummy queue poisoned").pop_front()
    }

    /// The number of transmitted frames waiting to be read.
    #[must_use]
    pub fn len(&self) -> usize {
        self.sent.lock().expect("dummy queue poisoned").len()
    }

    /// Returns `true` if no transmitted frames are waiting.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

/// Drains readiness bytes from the receive pipe, returning `true` if end-of-stream was reached
/// (the [`Injector`] was dropped).
pub(crate) fn drain_readiness(fd: BorrowedFd<'_>) -> bool {
    let mut buf = [0u8; 64];
    loop {
        // SAFETY: reading into a valid local buffer from a valid fd; the pipe is non-blocking.
        let n = unsafe { libc::read(fd.as_raw_fd(), buf.as_mut_ptr().cast(), buf.len()) };
        match n {
            0 => return true,             // all write ends closed: end of stream
            n if n < 0 => return false,   // EAGAIN (drained) or a transient error
            _ => {}                       // drained some bytes; keep going
        }
    }
}

pub(crate) fn make_pipe() -> Result<(OwnedFd, OwnedFd)> {
    let mut fds = [0 as RawFd; 2];
    // SAFETY: `fds` is a valid 2-element array that pipe2 fills with two fresh fds.
    let rc = unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) };
    if rc != 0 {
        return Err(crate::Error::Socket(io::Error::last_os_error()));
    }
    // SAFETY: pipe2 succeeded, so both descriptors are valid and owned.
    Ok(unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) })
}

/// Three fake interfaces (`dummy0`..`dummy2`) for tests that enumerate interfaces.
#[must_use]
pub fn interfaces() -> Vec<Interface> {
    (0..3)
        .map(|i| Interface {
            name: format!("dummy{i}"),
            index: IfIndex(i + 1),
            mac: Some(MacAddr([0x02, 0, 0, 0, 0, i as u8])),
            ips: Vec::new(),
            flags: 0,
            mtu: 1500,
        })
        .collect()
}