ferrix/crates/fib/src/lib.rs

//! Putting the best routes into the kernel's routing table, so that the
//! machine forwards packets by them (Linux, over rtnetlink).
//!
//! The routing table chooses a best route for each prefix. This crate
//! follows those choices and keeps the kernel's table the same: one route
//! for each prefix, to the best route's next hop. It also learns from the
//! kernel what interfaces and addresses the machine has, which is what
//! says whether a next hop can be reached at all.
//!
//! - [`netlink`] is the kernel's messages, as bytes and back.
//! - [`interfaces`] is the machine's interfaces and what is reachable on
//!   them.
//! - [`plan`] works out what to ask the kernel for.
//! - The rest, which only Linux has, asks it.
//!
//! # What is and is not touched
//!
//! Our routes are marked as BGP's ([`netlink::PROTOCOL_BGP`]) and carry
//! the configured metric, in the configured table. Only routes with that
//! mark, metric and table are ever replaced or removed; and all such
//! routes are taken to be ours, so two daemons must not share them.
//! Routes left behind by a run that ended badly are found at the next
//! start, and removed unless they are still wanted. A run that ends in
//! good order removes its routes.
//!
//! # What is installed
//!
//! A best route whose next hop is a neighbor: an address in the network
//! of one of the machine's interfaces. A route whose next hop is further
//! away than that, as with routes from a peer in our own AS that did not
//! set itself as next hop, would need the way to the next hop looked up in
//! turn, which is not done: the prefix then has no route in the kernel.

pub mod interfaces;
#[cfg(target_os = "linux")]
mod kernel;
pub mod netlink;
pub mod plan;

use std::net::IpAddr;
use std::sync::{Arc, Mutex, PoisonError};
use std::time::Duration;

use bgp_rib::table::RibTable;
use tokio::sync::watch;
use tokio::task::JoinHandle;

/// The metric our routes get unless configured otherwise: the one other
/// routing daemons give routes from external BGP peers, so that ours rank
/// against static and connected routes as an operator expects.
pub const DEFAULT_METRIC: u32 = 20;

/// How long routes left by an earlier run are kept unless configured
/// otherwise: time for sessions to come up and tables to be sent.
pub const DEFAULT_STALE_TIME: Duration = Duration::from_secs(120);

/// Where our routes go in the kernel.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FibConfig {
    /// The routing table: [`netlink::TABLE_MAIN`] to forward by them.
    pub table: u32,
    /// The metric, which with the table says which routes are ours.
    pub metric: u32,
    /// Have routes name nexthop objects where the kernel has them (Linux
    /// 5.3 and later), which makes installing many routes a good deal
    /// faster. Without, every route carries its own gateway.
    pub nexthop_objects: bool,
    /// How long routes of ours that an earlier run left in the kernel are
    /// kept after a start without the routing table having them. The
    /// table is empty until the peers have sent theirs again; removing
    /// everything at once would stop the machine forwarding meanwhile.
    /// After this long, what has not been announced again goes.
    pub stale_time: Duration,
}

impl Default for FibConfig {
    fn default() -> Self {
        Self {
            table: netlink::TABLE_MAIN,
            metric: DEFAULT_METRIC,
            nexthop_objects: true,
            stale_time: DEFAULT_STALE_TIME,
        }
    }
}

/// Why routes cannot be kept in the kernel.
#[derive(Debug, thiserror::Error)]
pub enum FibError {
    #[error("installing routes in the kernel is only supported on Linux")]
    Unsupported,
    #[error("the kernel's routing socket failed: {0}")]
    Io(#[from] std::io::Error),
    #[error("the kernel sent something that is not netlink: {0}")]
    Malformed(#[from] netlink::NetlinkError),
    #[error("the kernel refused to list {what}: error {errno}")]
    Dump { what: &'static str, errno: i32 },
    #[error(
        "not permitted to change the kernel's routes: the daemon needs the CAP_NET_ADMIN \
         capability (or root)"
    )]
    NotPermitted,
}

/// One of the machine's interfaces, for showing.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InterfaceStatus {
    pub index: u32,
    pub name: String,
    pub up: bool,
    /// Its addresses, each with the length of the network it is in.
    pub addresses: Vec<(IpAddr, u8)>,
}

/// How things stand with the kernel, for showing.
#[derive(Debug, Clone, Default)]
pub struct Status {
    /// Whether the kernel has been brought in step with the routing table
    /// at least once since the start.
    pub synced: bool,
    /// How many routes of ours the kernel has.
    pub installed: usize,
    /// How many of them are kept from an earlier run though the routing
    /// table does not have them (yet); see [`FibConfig::stale_time`].
    pub kept: usize,
    /// Prefixes whose best route is not in the kernel because its next
    /// hop is not usable, by why.
    pub unresolved: Vec<(&'static str, usize)>,
    /// Prefixes whose best route the kernel refused.
    pub refused: usize,
    /// How many requests the kernel has refused in all.
    pub errors: u64,
    /// The last such refusal.
    pub last_error: Option<String>,
    pub interfaces: Vec<InterfaceStatus>,
}

/// The machine's interfaces as the kernel last described them; `None`
/// until it has. Whoever holds one is told when they change.
pub type InterfacesWatch = watch::Receiver<Option<Arc<interfaces::Interfaces>>>;

/// How the task that keeps the kernel's routes is to end.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Stop {
    /// Taking our routes out of the kernel.
    Withdraw,
    /// Leaving them there, for the next run to find.
    Keep,
}

struct Shared {
    config: FibConfig,
    status: Mutex<Status>,
    interfaces: watch::Sender<Option<Arc<interfaces::Interfaces>>>,
    stop: watch::Sender<Option<Stop>>,
    /// How many routes of an earlier run were found in the kernel at the
    /// start, once that has been looked at.
    adopted: watch::Sender<Option<usize>>,
    /// Set to have what is kept of an earlier run go now.
    release: watch::Sender<bool>,
    /// Whether the task has ended.
    ended: watch::Sender<bool>,
}

/// Handle to the task that keeps the kernel's routes. Cheap to clone.
#[derive(Clone)]
pub struct Fib {
    shared: Arc<Shared>,
}

impl Fib {
    /// Starts keeping the kernel's routes in step with the best routes of
    /// `rib`, and returns a handle with the task that does it. The task
    /// ends only by failing, or when [`Fib::stop`] is called.
    ///
    /// Must be called from within a Tokio runtime. Fails at once where
    /// there is no kernel to talk to in this way.
    pub fn spawn(
        rib: Arc<RibTable>,
        config: FibConfig,
    ) -> Result<(Self, JoinHandle<Result<(), FibError>>), FibError> {
        let (stop, stopping) = watch::channel(None);
        let fib = Self {
            shared: Arc::new(Shared {
                config,
                status: Mutex::default(),
                interfaces: watch::channel(None).0,
                stop,
                adopted: watch::channel(None).0,
                release: watch::channel(false).0,
                ended: watch::channel(false).0,
            }),
        };
        let task = start(rib, fib.clone(), stopping)?;
        Ok((fib, task))
    }

    pub fn config(&self) -> FibConfig {
        self.shared.config
    }

    /// The machine's interfaces, for those who want to know whether a
    /// next hop can be reached before the kernel is asked to use it.
    pub fn interfaces(&self) -> InterfacesWatch {
        self.shared.interfaces.subscribe()
    }

    #[cfg_attr(not(target_os = "linux"), allow(dead_code))]
    fn publish_interfaces(&self, interfaces: &interfaces::Interfaces) {
        self.shared
            .interfaces
            .send_replace(Some(Arc::new(interfaces.clone())));
    }

    pub fn status(&self) -> Status {
        self.shared
            .status
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .clone()
    }

    /// Has the task take our routes out of the kernel and end, and waits
    /// until it has. Returns at once if it has ended already, for whatever
    /// reason; the task's own result says how.
    pub async fn stop(&self) {
        self.end(Stop::Withdraw).await;
    }

    /// Has the task stop following the routing table and end, leaving our
    /// routes in the kernel for the next run to find and keep for a while
    /// (see [`FibConfig::stale_time`]), and waits until it has. For a
    /// restart that is to go unnoticed by the packets being forwarded.
    pub async fn leave(&self) {
        self.end(Stop::Keep).await;
    }

    async fn end(&self, how: Stop) {
        let mut ended = self.shared.ended.subscribe();
        self.shared.stop.send_if_modified(|stop| {
            let first = stop.is_none();
            if first {
                *stop = Some(how);
            }
            first
        });
        // Fails only if the handle that holds the sender is gone, and this
        // is one.
        let _ = ended.wait_for(|ended| *ended).await;
    }

    /// How many routes of an earlier run the kernel had at the start, once
    /// the task has looked; zero if it ended before it did.
    pub async fn adopted(&self) -> usize {
        let mut adopted = self.shared.adopted.subscribe();
        let mut ended = self.shared.ended.subscribe();
        tokio::select! {
            found = adopted.wait_for(Option::is_some) => {
                found.ok().and_then(|found| *found).unwrap_or(0)
            }
            _ = ended.wait_for(|ended| *ended) => 0,
        }
    }

    /// Has the routes of an earlier run that the routing table does not
    /// have go now, rather than when [`FibConfig::stale_time`] is up: the
    /// peers have all sent their routes, and what is not among them is no
    /// longer anybody's.
    pub fn release_kept(&self) {
        self.shared.release.send_replace(true);
    }

    #[cfg_attr(not(target_os = "linux"), allow(dead_code))]
    fn update(&self, change: impl FnOnce(&mut Status)) {
        change(
            &mut self
                .shared
                .status
                .lock()
                .unwrap_or_else(PoisonError::into_inner),
        );
    }
}

#[cfg(target_os = "linux")]
fn start(
    rib: Arc<RibTable>,
    fib: Fib,
    stopping: watch::Receiver<Option<Stop>>,
) -> Result<JoinHandle<Result<(), FibError>>, FibError> {
    // Opened here, so that a machine that cannot is known of at once.
    let sockets = kernel::Sockets::open()?;
    Ok(tokio::spawn(async move {
        let result = kernel::run(rib, fib.clone(), sockets, stopping).await;
        fib.shared.ended.send_replace(true);
        result
    }))
}

#[cfg(not(target_os = "linux"))]
fn start(
    _rib: Arc<RibTable>,
    _fib: Fib,
    _stopping: watch::Receiver<Option<Stop>>,
) -> Result<JoinHandle<Result<(), FibError>>, FibError> {
    Err(FibError::Unsupported)
}