Install best routes in the Linux kernel over netlink

New crate bgp-fib. It follows the routing table's best routes and keeps
the kernel's routing table the same, and learns the machine's interfaces
and addresses from the kernel, which is what says whether a next hop can
be reached.

- netlink: the rtnetlink messages used (routes, addresses, links, errors),
  as bytes and back. No I/O; property tested and fuzzed.
- interfaces: what is directly reachable, and from which interface.
- plan: what to ask the kernel for, given the best routes and what the
  kernel has. No I/O; tested against a model kernel.
- kernel (Linux only): the sockets and the task. Requests go in batches;
  routes of ours found at startup are kept if still right and removed if
  not; a change of interfaces has every route worked out again; a clean
  stop removes our routes.

Only routes marked proto bgp with the configured table and metric are
ever touched. A best route is installed if its next hop is a neighbor;
next hops further away are not resolved through other routes yet, and
such prefixes are counted and shown instead.

Configured with `kernel { table metric }`; off by default. The daemon
needs CAP_NET_ADMIN and stops if the kernel refuses it. On shutdown it
closes sessions and then removes its routes.

/api/kernel and the status page show installed, unresolved and refused
counts, the last error and the interfaces; five bgp_kernel_* metrics.

Testing: scripts/kernel-tests.sh runs, in network namespaces of their
own and with only the capabilities the daemon needs, the installer
against `ip route`, the daemon with a peer, and two daemons that route
for each other with pings across over IPv4 and IPv6.
scripts/linux-vm.sh runs that, or anything else, in a Lima VM from
macOS. CI runs the kernel tests and fuzzes the netlink decoder.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Rayhaan JaufeerallyandClaude Fable 5.1 committed 2026-10-02 15:15:43 +02:00
1 parent 1f84235e87
commit 1fc44798b7
38 files changed
+3745 -16

No files matched your search

+29
View File
@@ -0,0 +1,29 @@
[package]
name = "bgp-fib"
authors.workspace = true
edition.workspace = true
homepage.workspace = true
license.workspace = true
repository.workspace = true
version.workspace = true
[dependencies]
bgp-packet.workspace = true
bgp-primitives.workspace = true
bgp-rib.workspace = true
bytes.workspace = true
prometheus.workspace = true
thiserror.workspace = true
tokio = { workspace = true, features = ["full"] }
tracing.workspace = true
# The socket to the kernel. Only Linux has it; everything else in this
# crate builds, and is tested, anywhere.
[target.'cfg(target_os = "linux")'.dependencies]
netlink-sys = { version = "0.8", features = ["tokio_socket"] }
[dev-dependencies]
proptest.workspace = true
[lints]
workspace = true
+231
View File
@@ -0,0 +1,231 @@
//! The machine's interfaces and their addresses, as the kernel reports
//! them, and what they say about a next hop: whether it can be reached
//! directly, and from which interface.
use std::collections::{HashMap, HashSet};
use std::net::IpAddr;
use bgp_primitives::constants::IpVersion;
use bgp_primitives::ip_prefix::IpPrefix;
use crate::netlink::{Address, Link, Message};
/// The interface a neighbor is reached on.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Connected {
/// The interface's index.
pub interface: u32,
/// Our own address in the network the neighbor is in.
pub local: IpAddr,
}
/// The interfaces there are and the addresses they have.
#[derive(Debug, Clone, Default)]
pub struct Interfaces {
links: HashMap<u32, Link>,
addresses: HashSet<Address>,
}
/// `address` as the prefix that is only it.
fn host(address: IpAddr) -> IpPrefix {
match address {
IpAddr::V4(v4) => IpPrefix::new(IpVersion::Ipv4, v4.octets(), 32),
IpAddr::V6(v6) => IpPrefix::new(IpVersion::Ipv6, v6.octets(), 128),
}
.expect("an address is a prefix of its own length")
}
impl Interfaces {
/// Takes in what a message from the kernel says about links and
/// addresses. Returns whether anything is different for it.
pub fn apply(&mut self, message: &Message) -> bool {
match message {
Message::Link {
present: true,
link,
} => self.links.insert(link.index, link.clone()).as_ref() != Some(link),
Message::Link {
present: false,
link,
} => {
self.addresses
.retain(|address| address.interface != link.index);
self.links.remove(&link.index).is_some()
}
Message::Address {
present: true,
address,
} => self.addresses.insert(*address),
Message::Address {
present: false,
address,
} => self.addresses.remove(address),
_ => false,
}
}
/// The interfaces, by index.
pub fn links(&self) -> impl Iterator<Item = &Link> {
self.links.values()
}
pub fn link(&self, index: u32) -> Option<&Link> {
self.links.get(&index)
}
/// The addresses of the interface with this index.
pub fn addresses_of(&self, interface: u32) -> impl Iterator<Item = &Address> {
self.addresses
.iter()
.filter(move |address| address.interface == interface)
}
/// Whether `address` is one of this machine's own.
pub fn is_local(&self, address: IpAddr) -> bool {
self.addresses.iter().any(|ours| ours.local == address)
}
/// Where `neighbor` can be reached without a gateway: the interface
/// that is up and has an address in a network `neighbor` is in, the
/// most specific such network if there are several. `None` if there is
/// none, or if `neighbor` is this machine itself.
pub fn connected(&self, neighbor: IpAddr) -> Option<Connected> {
if self.is_local(neighbor) {
return None;
}
let neighbor = host(neighbor);
self.addresses
.iter()
.filter(|address| address.network.contains(&neighbor))
.filter(|address| {
self.links
.get(&address.interface)
.is_some_and(|link| link.up)
})
.max_by_key(|address| address.network.length())
.map(|address| Connected {
interface: address.interface,
local: address.local,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
fn link(index: u32, name: &str, up: bool) -> Message {
Message::Link {
present: true,
link: Link {
index,
name: name.into(),
up,
},
}
}
fn address(interface: u32, local: &str, network: &str) -> Address {
Address {
interface,
local: local.parse().unwrap(),
network: IpPrefix::try_from(network).unwrap(),
}
}
fn present(address: Address) -> Message {
Message::Address {
present: true,
address,
}
}
fn some() -> Interfaces {
let mut interfaces = Interfaces::default();
for message in [
link(1, "lo", true),
link(2, "eth0", true),
link(3, "eth1", true),
link(4, "eth2", false),
present(address(1, "127.0.0.1", "127.0.0.0/8")),
present(address(2, "192.0.2.1", "192.0.2.0/24")),
present(address(2, "2001:db8::1", "2001:db8::/64")),
// A more specific network inside eth0's, on eth1.
present(address(3, "192.0.2.129", "192.0.2.128/25")),
present(address(4, "198.51.100.1", "198.51.100.0/24")),
] {
assert!(interfaces.apply(&message));
}
interfaces
}
fn connected(interfaces: &Interfaces, neighbor: &str) -> Option<(u32, String)> {
interfaces
.connected(neighbor.parse().unwrap())
.map(|c| (c.interface, c.local.to_string()))
}
#[test]
fn a_neighbor_is_reached_from_the_interface_whose_network_it_is_in() {
let interfaces = some();
assert_eq!(
connected(&interfaces, "192.0.2.7"),
Some((2, "192.0.2.1".into()))
);
assert_eq!(
connected(&interfaces, "2001:db8::7"),
Some((2, "2001:db8::1".into()))
);
// The most specific network wins.
assert_eq!(
connected(&interfaces, "192.0.2.200"),
Some((3, "192.0.2.129".into()))
);
// Not in any network of ours.
assert_eq!(connected(&interfaces, "203.0.113.1"), None);
assert_eq!(connected(&interfaces, "2001:db8:1::1"), None);
// In the network of an interface that is down.
assert_eq!(connected(&interfaces, "198.51.100.7"), None);
// Ourselves.
assert_eq!(connected(&interfaces, "192.0.2.1"), None);
assert!(interfaces.is_local("192.0.2.1".parse().unwrap()));
assert!(!interfaces.is_local("192.0.2.7".parse().unwrap()));
}
#[test]
fn what_the_kernel_reports_changes_what_is_reachable() {
let mut interfaces = some();
// Told again what is known: nothing changes.
assert!(!interfaces.apply(&link(2, "eth0", true)));
assert!(!interfaces.apply(&present(address(2, "192.0.2.1", "192.0.2.0/24"))));
// An interface comes up.
assert!(interfaces.apply(&link(4, "eth2", true)));
assert_eq!(
connected(&interfaces, "198.51.100.7"),
Some((4, "198.51.100.1".into()))
);
// An address goes.
assert!(interfaces.apply(&Message::Address {
present: false,
address: address(3, "192.0.2.129", "192.0.2.128/25"),
}));
assert_eq!(
connected(&interfaces, "192.0.2.200"),
Some((2, "192.0.2.1".into()))
);
// An interface goes, and its addresses with it.
assert!(interfaces.apply(&Message::Link {
present: false,
link: Link {
index: 2,
name: "eth0".into(),
up: true
},
}));
assert_eq!(connected(&interfaces, "192.0.2.7"), None);
assert_eq!(interfaces.addresses_of(2).count(), 0);
// What is not about interfaces changes nothing.
assert!(!interfaces.apply(&Message::Done));
}
}
+566
View File
@@ -0,0 +1,566 @@
//! Talking to the kernel: the sockets, and the task that keeps the
//! kernel's routes in step with the routing table's best routes.
use std::collections::VecDeque;
use std::io;
use std::net::IpAddr;
use std::sync::{Arc, LazyLock};
use std::time::Duration;
use bgp_packet::message::NlriNextHop;
use bgp_primitives::constants::IpVersion;
use bgp_primitives::ip_prefix::IpPrefix;
use bgp_rib::table::{Event, PeerIndex, Reader, RibTable};
use bytes::BytesMut;
use netlink_sys::protocols::NETLINK_ROUTE;
use netlink_sys::{AsyncSocket, AsyncSocketExt, SocketAddr, TokioSocket};
use prometheus::{
IntCounter, IntCounterVec, IntGauge, register_int_counter, register_int_counter_vec,
register_int_gauge,
};
use tokio::sync::watch;
use tokio::time::{Instant, sleep_until};
use tracing::{debug, info, warn};
use crate::interfaces::Interfaces;
use crate::netlink::{
self, Change, Dump, KIND_UNICAST, Message, PROTOCOL_BGP, Route, decode, group,
};
use crate::plan::{Op, Plan, Target, resolve};
use crate::{Fib, FibConfig, FibError, InterfaceStatus};
/// How many best routes are taken from the routing table, and how many
/// requests are sent to the kernel, at a time.
const BATCH: usize = 256;
/// The most a read from the kernel can return: it sends up to 32 KiB at a
/// time in answer to a dump.
const READ_BUFFER: usize = 256 * 1024;
/// How long after the machine's interfaces or addresses change the routes
/// are worked out again. Such changes come in bursts.
const SETTLE: Duration = Duration::from_secs(1);
/// The errors of the kernel that have a meaning here.
mod errno {
pub const EPERM: i32 = 1;
pub const ENOENT: i32 = 2;
pub const ESRCH: i32 = 3;
}
static ROUTES: LazyLock<IntGauge> = LazyLock::new(|| {
register_int_gauge!(
"bgp_kernel_routes",
"Routes of ours in the kernel's routing table"
)
.expect("metric registration")
});
static UNRESOLVED: LazyLock<IntGauge> = LazyLock::new(|| {
register_int_gauge!(
"bgp_kernel_routes_unresolved",
"Prefixes whose best route is not in the kernel because its next hop is not usable"
)
.expect("metric registration")
});
static REFUSED: LazyLock<IntGauge> = LazyLock::new(|| {
register_int_gauge!(
"bgp_kernel_routes_refused",
"Prefixes whose best route the kernel refused to take"
)
.expect("metric registration")
});
static CHANGES: LazyLock<IntCounterVec> = LazyLock::new(|| {
register_int_counter_vec!(
"bgp_kernel_route_changes_total",
"Changes to its routing table the kernel was asked for, by kind",
&["change"]
)
.expect("metric registration")
});
static ERRORS: LazyLock<IntCounter> = LazyLock::new(|| {
register_int_counter!("bgp_kernel_errors_total", "Requests the kernel refused")
.expect("metric registration")
});
/// A routing socket.
struct Socket {
socket: TokioSocket,
sequence: u32,
buf: BytesMut,
}
impl Socket {
/// A socket that is sent the notifications of `groups`.
fn open(groups: u32) -> io::Result<Self> {
let mut socket = TokioSocket::new(NETLINK_ROUTE)?;
socket.socket_mut().bind(&SocketAddr::new(0, groups))?;
Ok(Self {
socket,
sequence: 0,
buf: BytesMut::with_capacity(READ_BUFFER),
})
}
fn next_sequence(&mut self) -> u32 {
self.sequence = self.sequence.wrapping_add(1);
self.sequence
}
/// Waits for the kernel to send something, and returns it.
async fn read(&mut self) -> io::Result<&[u8]> {
self.buf.clear();
self.buf.reserve(READ_BUFFER);
self.socket.recv(&mut self.buf).await?;
Ok(&self.buf)
}
/// Asks for all of `what` and gives each message of the answer to
/// `each`.
async fn dump(&mut self, what: Dump, mut each: impl FnMut(Message)) -> Result<(), FibError> {
let sequence = self.next_sequence();
let mut request = BytesMut::new();
netlink::dump_request(&mut request, sequence, what);
self.socket.send(&request).await?;
loop {
let mut rest = self.read().await?;
while !rest.is_empty() {
let (header, message, used) = decode(rest)?;
rest = &rest[used..];
if header.sequence != sequence {
continue;
}
match message {
Message::Done => return Ok(()),
Message::Outcome { errno } if errno != 0 => {
return Err(FibError::Dump {
what: match what {
Dump::Routes(_) => "its routes",
Dump::Addresses => "its addresses",
Dump::Links => "its interfaces",
},
errno,
});
}
message => each(message),
}
}
}
}
/// Asks the kernel for `requests`, each a change to a route, and
/// returns which of them it refused: the place in `requests` and the
/// error's number.
async fn change(
&mut self,
requests: &[(Change, Route)],
) -> Result<Vec<(usize, i32)>, FibError> {
let mut refused = Vec::new();
for batch_start in (0..requests.len()).step_by(BATCH) {
let batch = &requests[batch_start..requests.len().min(batch_start + BATCH)];
// One write for all of them. The kernel answers a request only
// if it fails, except the last, which it answers either way:
// that answer says the batch is done.
let first = self.sequence.wrapping_add(1);
let mut wire = BytesMut::with_capacity(batch.len() * 80);
for (at, (change, route)) in batch.iter().enumerate() {
let sequence = self.next_sequence();
netlink::route_request(&mut wire, sequence, *change, at + 1 == batch.len(), route);
}
let last = self.sequence;
self.socket.send(&wire).await?;
let mut done = false;
while !done {
let mut rest = self.read().await?;
while !rest.is_empty() {
let (header, message, used) = decode(rest)?;
rest = &rest[used..];
let at = header.sequence.wrapping_sub(first) as usize;
let Message::Outcome { errno } = message else {
continue;
};
if at >= batch.len() {
continue;
}
if errno != 0 {
refused.push((batch_start + at, errno));
}
done |= header.sequence == last;
}
}
}
Ok(refused)
}
}
/// The sockets the task uses.
pub(crate) struct Sockets {
/// For asking, and the answers.
requests: Socket,
/// For what the kernel tells unasked: changes to the interfaces and
/// their addresses.
news: Socket,
}
impl Sockets {
pub(crate) fn open() -> Result<Self, FibError> {
Ok(Self {
requests: Socket::open(0)?,
news: Socket::open(group::LINK | group::IPV4_ADDRESS | group::IPV6_ADDRESS)?,
})
}
}
/// A change to the best routes, as taken out of the routing table to be
/// dealt with once the table is let go of.
enum News {
Reset,
Best {
prefix: IpPrefix,
peer: PeerIndex,
next_hop: Option<NlriNextHop>,
},
Withdrawn(IpPrefix),
Synced,
}
struct Task {
rib: Arc<RibTable>,
fib: Fib,
config: FibConfig,
requests: Socket,
interfaces: Interfaces,
plan: Plan,
synced: bool,
}
impl Task {
fn route(&self, op: &Op) -> (Change, Route) {
let (change, target) = match op {
Op::Install { target, .. } => (Change::Replace, Some(target)),
Op::Remove { .. } => (Change::Remove, None),
};
let route = Route {
prefix: *op.prefix(),
table: self.config.table,
protocol: PROTOCOL_BGP,
metric: self.config.metric,
gateway: target.map(|target| target.gateway),
interface: target.map(|target| target.interface),
kind: KIND_UNICAST,
};
(change, route)
}
/// Whether a route the kernel has is one of ours.
fn is_ours(&self, route: &Route) -> bool {
route.protocol == PROTOCOL_BGP
&& route.table == self.config.table
&& route.metric == self.config.metric
}
/// Learns the machine's interfaces and addresses anew.
async fn learn_interfaces(&mut self) -> Result<(), FibError> {
let mut interfaces = Interfaces::default();
for what in [Dump::Links, Dump::Addresses] {
self.requests
.dump(what, |message| {
interfaces.apply(&message);
})
.await?;
}
self.interfaces = interfaces;
Ok(())
}
/// Finds the routes of ours that the kernel has from before.
async fn adopt_routes(&mut self) -> Result<(), FibError> {
let mut found = Vec::new();
for version in [IpVersion::Ipv4, IpVersion::Ipv6] {
self.requests
.dump(Dump::Routes(version), |message| {
if let Message::Route {
present: true,
route,
} = message
{
found.push(route);
}
})
.await?;
}
for route in found {
if !self.is_ours(&route) {
continue;
}
let target = match (route.gateway, route.interface) {
(Some(gateway), Some(interface)) if route.kind == KIND_UNICAST => {
Some(Target { gateway, interface })
}
_ => None,
};
self.plan.adopt(route.prefix, target);
}
if self.plan.installed() > 0 {
info!(
routes = self.plan.installed(),
"Found routes of ours in the kernel from an earlier run"
);
}
Ok(())
}
/// Asks the kernel for `ops` and deals with what it refuses.
async fn apply(&mut self, ops: Vec<Op>) -> Result<(), FibError> {
let mut pending = VecDeque::from([ops]);
while let Some(ops) = pending.pop_front() {
if ops.is_empty() {
continue;
}
let requests: Vec<(Change, Route)> = ops.iter().map(|op| self.route(op)).collect();
for (change, _) in &requests {
let label = match change {
Change::Replace => "install",
Change::Remove => "remove",
};
CHANGES.with_label_values(&[label]).inc();
}
let mut instead = Vec::new();
for (at, errno) in self.requests.change(&requests).await? {
let op = &ops[at];
if errno == errno::EPERM {
return Err(FibError::NotPermitted);
}
// Removing what is not there is as good as done.
if matches!(op, Op::Remove { .. }) && matches!(errno, errno::ESRCH | errno::ENOENT)
{
continue;
}
let error = io::Error::from_raw_os_error(errno);
let what = match op {
Op::Install { prefix, target } => format!(
"install {prefix} via {} on {}",
target.gateway,
self.interface_name(target.interface)
),
Op::Remove { prefix } => format!("remove {prefix}"),
};
// Counted always, said once in a while: a table's worth of
// refusals is a table's worth of lines.
ERRORS.inc();
let said = format!("the kernel refused to {what}: {error}");
debug!("{said}");
self.fib.update(|status| {
if status.errors == 0 || status.errors.is_power_of_two() {
warn!(refusals = status.errors + 1, "{said}");
}
status.errors += 1;
status.last_error = Some(said);
});
instead.extend(self.plan.refused_by_kernel(op));
}
pending.push_back(instead);
}
Ok(())
}
fn interface_name(&self, index: u32) -> String {
self.interfaces
.link(index)
.map_or_else(|| format!("interface {index}"), |link| link.name.clone())
}
/// Takes the next changes of best route from the routing table and
/// brings the kernel in step with them.
async fn follow(&mut self, reader: &mut Reader) -> Result<(), FibError> {
let mut news = Vec::new();
// Only what is needed is taken while the table is held.
reader.poll(BATCH, |event| {
news.push(match event {
Event::Reset => News::Reset,
Event::Synced => News::Synced,
Event::Best {
prefix,
peer,
attrs,
..
} => News::Best {
prefix,
peer,
next_hop: attrs.next_hop(),
},
Event::Withdrawn { prefix } => News::Withdrawn(prefix),
// Not given to a follower of the best routes.
Event::Routes { .. } => return,
});
});
let mut ops = Vec::new();
for item in news {
match item {
News::Reset => self.plan.reset(),
News::Synced => {
let stale = self.plan.synced();
if !self.synced || !stale.is_empty() {
info!(
installed = self.plan.installed(),
unresolved = self.plan.unresolved(),
removed_stale = stale.len(),
"The kernel's routes are in step with the routing table"
);
}
self.synced = true;
ops.extend(stale);
}
News::Best {
prefix,
peer,
next_hop,
} => {
// Whose link a link-local next hop is on is asked only
// where there is one.
let peer_address =
|rib: &RibTable| -> Option<IpAddr> { Some(rib.peer(peer)?.address) };
let needs_peer = matches!(
next_hop,
Some(NlriNextHop::Ipv6(_) | NlriNextHop::Ipv6WithLl { .. })
);
let peer = needs_peer.then(|| peer_address(&self.rib)).flatten();
let target = resolve(&self.interfaces, &prefix, next_hop.as_ref(), peer);
ops.extend(self.plan.best(prefix, target));
}
News::Withdrawn(prefix) => ops.extend(self.plan.withdrawn(prefix)),
}
}
self.apply(ops).await
}
/// Says how things stand, for whoever asks.
fn publish(&self, interfaces_changed: bool) {
ROUTES.set(self.plan.installed() as i64);
UNRESOLVED.set(self.plan.unresolved() as i64);
REFUSED.set(self.plan.refused() as i64);
let interfaces = interfaces_changed.then(|| {
let mut interfaces: Vec<InterfaceStatus> = self
.interfaces
.links()
.map(|link| {
let mut addresses: Vec<(IpAddr, u8)> = self
.interfaces
.addresses_of(link.index)
.map(|address| (address.local, address.network.length()))
.collect();
addresses.sort();
InterfaceStatus {
index: link.index,
name: link.name.clone(),
up: link.up,
addresses,
}
})
.collect();
interfaces.sort_by_key(|interface| interface.index);
interfaces
});
self.fib.update(|status| {
status.synced = self.synced;
status.installed = self.plan.installed();
status.refused = self.plan.refused();
status.unresolved = self
.plan
.unresolved_by_reason()
.map(|(reason, count)| (reason.as_str(), count))
.collect();
if let Some(interfaces) = interfaces {
status.interfaces = interfaces;
}
});
}
}
/// Keeps the kernel's routes in step with the best routes of `rib` until
/// told to stop, and then takes them out.
pub(crate) async fn run(
rib: Arc<RibTable>,
fib: Fib,
sockets: Sockets,
mut stopping: watch::Receiver<bool>,
) -> Result<(), FibError> {
let Sockets { requests, mut news } = sockets;
let mut task = Task {
config: fib.config(),
rib,
fib,
requests,
interfaces: Interfaces::default(),
plan: Plan::default(),
synced: false,
};
// The socket for news was bound before this: what changes while the
// interfaces are being learned is told there.
task.learn_interfaces().await?;
task.adopt_routes().await?;
task.publish(true);
info!(
table = task.config.table,
metric = task.config.metric,
interfaces = task.interfaces.links().count(),
"Installing best routes in the kernel"
);
// A reader is first given the whole table, which is what brings the
// kernel in step; starting a new one does it over.
let mut reader = task.rib.reader();
let mut again_at: Option<Instant> = None;
loop {
tokio::select! {
() = reader.changed() => {
task.follow(&mut reader).await?;
task.publish(false);
}
read = news.read() => {
let changed = match read {
Ok(mut rest) => {
let mut changed = false;
while !rest.is_empty() {
let (_, message, used) = decode(rest)?;
rest = &rest[used..];
changed |= task.interfaces.apply(&message);
}
changed
}
// The kernel had more to say than the socket had room
// for (ENOBUFS): what was missed is asked for.
Err(e) => {
warn!(err = %e, "Missed news from the kernel; asking for the interfaces again");
task.learn_interfaces().await?;
true
}
};
if changed {
// What a next hop resolves to may be different now,
// for any route.
again_at = Some(Instant::now() + SETTLE);
task.publish(true);
}
}
() = async { sleep_until(again_at.expect("checked by the guard")).await },
if again_at.is_some() =>
{
again_at = None;
info!("The machine's interfaces changed; working the kernel's routes out again");
reader = task.rib.reader();
}
_ = stopping.changed() => {
let ops = task.plan.clear();
info!(routes = ops.len(), "Taking our routes out of the kernel");
task.apply(ops).await?;
task.synced = false;
task.publish(false);
return Ok(());
}
}
}
}
+213
View File
@@ -0,0 +1,213 @@
//! 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 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;
/// 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,
}
impl Default for FibConfig {
fn default() -> Self {
Self {
table: netlink::TABLE_MAIN,
metric: DEFAULT_METRIC,
}
}
}
/// 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,
/// 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>,
}
struct Shared {
config: FibConfig,
status: Mutex<Status>,
stop: 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(false);
let fib = Self {
shared: Arc::new(Shared {
config,
status: Mutex::default(),
stop,
ended: watch::channel(false).0,
}),
};
let task = start(rib, fib.clone(), stopping)?;
Ok((fib, task))
}
pub fn config(&self) -> FibConfig {
self.shared.config
}
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) {
let mut ended = self.shared.ended.subscribe();
self.shared.stop.send_replace(true);
// Fails only if the handle that holds the sender is gone, and this
// is one.
let _ = ended.wait_for(|ended| *ended).await;
}
#[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<bool>,
) -> 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<bool>,
) -> Result<JoinHandle<Result<(), FibError>>, FibError> {
Err(FibError::Unsupported)
}
+836
View File
@@ -0,0 +1,836 @@
//! The messages of the kernel's routing socket (rtnetlink), as bytes and
//! back: the part of them this crate uses. No I/O here.
//!
//! Every message begins with a header of sixteen octets (`nlmsghdr`: length,
//! type, flags, sequence number, sender), then a fixed part that depends on
//! the type, then attributes: each a length, a type and a value, padded to
//! four octets. Numbers are in the byte order of the machine, the kernel
//! being on the same one; addresses are in network order as everywhere.
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use bgp_primitives::constants::IpVersion;
use bgp_primitives::ip_prefix::IpPrefix;
use bytes::{BufMut, BytesMut};
/// The routing protocol that routes of ours are marked as having come
/// from (`RTPROT_BGP`). Routes with another mark are never touched.
pub const PROTOCOL_BGP: u8 = 186;
/// The table that forwarding uses unless rules say otherwise
/// (`RT_TABLE_MAIN`).
pub const TABLE_MAIN: u32 = 254;
/// A route that forwards to a gateway or out of an interface
/// (`RTN_UNICAST`), as opposed to one that drops, say.
pub const KIND_UNICAST: u8 = 1;
const HEADER_LEN: usize = 16;
mod message_type {
pub const ERROR: u16 = 2;
pub const DONE: u16 = 3;
pub const NEW_LINK: u16 = 16;
pub const DEL_LINK: u16 = 17;
pub const GET_LINK: u16 = 18;
pub const NEW_ADDRESS: u16 = 20;
pub const DEL_ADDRESS: u16 = 21;
pub const GET_ADDRESS: u16 = 22;
pub const NEW_ROUTE: u16 = 24;
pub const DEL_ROUTE: u16 = 25;
pub const GET_ROUTE: u16 = 26;
}
mod flag {
pub const REQUEST: u16 = 0x1;
pub const ACK: u16 = 0x4;
/// For a request to read: all of them (`NLM_F_ROOT | NLM_F_MATCH`).
pub const DUMP: u16 = 0x300;
/// For a request to add: take the place of the one there is.
pub const REPLACE: u16 = 0x100;
/// For a request to add: make it if there is none.
pub const CREATE: u16 = 0x400;
}
/// The groups of notifications a socket can ask for when it is bound
/// (`RTMGRP_*`).
pub mod group {
pub const LINK: u32 = 0x1;
pub const IPV4_ADDRESS: u32 = 0x10;
pub const IPV6_ADDRESS: u32 = 0x100;
}
const AF_INET: u8 = 2;
const AF_INET6: u8 = 10;
/// `RT_TABLE_COMPAT`: the table's number is in an attribute, being too
/// large for the octet the fixed part has for it.
const TABLE_IN_ATTRIBUTE: u8 = 252;
const SCOPE_UNIVERSE: u8 = 0;
mod route_attribute {
pub const DESTINATION: u16 = 1;
pub const OUT_INTERFACE: u16 = 4;
pub const GATEWAY: u16 = 5;
pub const PRIORITY: u16 = 6;
pub const TABLE: u16 = 15;
}
mod address_attribute {
pub const ADDRESS: u16 = 1;
pub const LOCAL: u16 = 2;
}
mod link_attribute {
pub const NAME: u16 = 3;
}
const INTERFACE_UP: u32 = 0x1;
const INTERFACE_RUNNING: u32 = 0x40;
/// A route in one of the kernel's tables.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct Route {
pub prefix: IpPrefix,
pub table: u32,
/// Who put it there; see [`PROTOCOL_BGP`].
pub protocol: u8,
/// The kernel's "priority": of several routes for a prefix, the one
/// with the lowest is used.
pub metric: u32,
pub gateway: Option<IpAddr>,
/// The index of the interface packets leave by.
pub interface: Option<u32>,
/// See [`KIND_UNICAST`].
pub kind: u8,
}
/// An address of one of the machine's interfaces.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct Address {
/// The index of the interface.
pub interface: u32,
/// The address itself.
pub local: IpAddr,
/// The network it is in: what can be reached on the interface without
/// a gateway.
pub network: IpPrefix,
}
/// One of the machine's interfaces.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Link {
pub index: u32,
pub name: String,
/// Whether it is switched on and working.
pub up: bool,
}
/// What every message begins with, as far as it matters here.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Header {
pub flags: u16,
/// For an answer, the sequence number of the request it answers.
pub sequence: u32,
}
/// A message from the kernel.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Message {
/// A route there is (in answer to a dump, or newly) or was.
Route {
present: bool,
route: Route,
},
Address {
present: bool,
address: Address,
},
Link {
present: bool,
link: Link,
},
/// What became of a request: `errno` is zero if it was done, and
/// otherwise the error's number.
Outcome {
errno: i32,
},
/// The answer to a dump is complete.
Done,
/// Something this crate has no use for.
Other {
message_type: u16,
},
}
/// Bytes that are not a netlink message.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum NetlinkError {
#[error("a message of {claimed} octets in {available}")]
Length { claimed: usize, available: usize },
#[error("a message of type {message_type} too short for what it is")]
Truncated { message_type: u16 },
}
fn ne_u16(bytes: &[u8]) -> u16 {
u16::from_ne_bytes([bytes[0], bytes[1]])
}
fn ne_u32(bytes: &[u8]) -> u32 {
u32::from_ne_bytes([bytes[0], bytes[1], bytes[2], bytes[3]])
}
/// Rounds a length up to the four octets everything is aligned to.
fn aligned(length: usize) -> usize {
(length + 3) & !3
}
/// The attributes in `bytes`: each its type and value. Stops at what is
/// not an attribute.
fn attributes(mut bytes: &[u8]) -> impl Iterator<Item = (u16, &[u8])> {
std::iter::from_fn(move || {
let header = bytes.first_chunk::<4>()?;
let length = usize::from(ne_u16(header));
let kind = ne_u16(&header[2..]);
let value = bytes.get(4..length)?;
bytes = bytes.get(aligned(length)..).unwrap_or(&[]);
Some((kind, value))
})
}
fn version_of(family: u8) -> Option<IpVersion> {
match family {
AF_INET => Some(IpVersion::Ipv4),
AF_INET6 => Some(IpVersion::Ipv6),
_ => None,
}
}
fn family_of(version: IpVersion) -> u8 {
match version {
IpVersion::Ipv4 => AF_INET,
IpVersion::Ipv6 => AF_INET6,
}
}
/// An address of `version` from exactly its octets.
fn address(version: IpVersion, bytes: &[u8]) -> Option<IpAddr> {
match version {
IpVersion::Ipv4 => Some(Ipv4Addr::from(<[u8; 4]>::try_from(bytes).ok()?).into()),
IpVersion::Ipv6 => Some(Ipv6Addr::from(<[u8; 16]>::try_from(bytes).ok()?).into()),
}
}
/// Reads the message at the front of `buf`, which is what a read from the
/// socket returned, or what is left of it: the header, the message, and
/// how many octets it took. A read can hold several messages.
pub fn decode(buf: &[u8]) -> Result<(Header, Message, usize), NetlinkError> {
let bad_length = |claimed| NetlinkError::Length {
claimed,
available: buf.len(),
};
let header = buf
.first_chunk::<HEADER_LEN>()
.ok_or(bad_length(HEADER_LEN))?;
let length = ne_u32(header) as usize;
let message_type = ne_u16(&header[4..]);
let body = buf
.get(HEADER_LEN..length)
.ok_or_else(|| bad_length(length))?;
let header = Header {
flags: ne_u16(&header[6..]),
sequence: ne_u32(&header[8..]),
};
let truncated = NetlinkError::Truncated { message_type };
let other = Message::Other { message_type };
let message = match message_type {
message_type::ERROR => Message::Outcome {
// The kernel gives the error's number negated.
errno: body
.first_chunk::<4>()
.map(|errno| i32::from_ne_bytes(*errno).wrapping_neg())
.ok_or(truncated)?,
},
message_type::DONE => Message::Done,
message_type::NEW_ROUTE | message_type::DEL_ROUTE => {
// rtmsg: family, destination length, source length, TOS,
// table, protocol, scope, type, and four octets of flags.
let (fixed, rest) = body.split_at_checked(12).ok_or(truncated)?;
match route(fixed, rest) {
Some(route) => Message::Route {
present: message_type == message_type::NEW_ROUTE,
route,
},
None => other,
}
}
message_type::NEW_ADDRESS | message_type::DEL_ADDRESS => {
// ifaddrmsg: family, prefix length, flags, scope, and the
// interface's index.
let (fixed, rest) = body.split_at_checked(8).ok_or(truncated)?;
match interface_address(fixed, rest) {
Some(address) => Message::Address {
present: message_type == message_type::NEW_ADDRESS,
address,
},
None => other,
}
}
message_type::NEW_LINK | message_type::DEL_LINK => {
// ifinfomsg: family, padding, device type, index, flags, and
// which flags changed.
let (fixed, rest) = body.split_at_checked(16).ok_or(truncated)?;
let flags = ne_u32(&fixed[8..]);
let name = attributes(rest)
.find(|(kind, _)| *kind == link_attribute::NAME)
.map(|(_, value)| {
// The name ends with a zero octet.
let name = value.split(|octet| *octet == 0).next().unwrap_or(&[]);
String::from_utf8_lossy(name).into_owned()
})
.unwrap_or_default();
Message::Link {
present: message_type == message_type::NEW_LINK,
link: Link {
index: ne_u32(&fixed[4..]),
name,
up: flags & INTERFACE_UP != 0 && flags & INTERFACE_RUNNING != 0,
},
}
}
_ => other,
};
// The next message begins where this one ends, rounded up; the last
// need not be padded.
Ok((header, message, aligned(length).min(buf.len())))
}
/// A route from the fixed part of its message and its attributes. `None`
/// if it is of a family, or a shape, this crate has no notion of.
fn route(fixed: &[u8], attributes_of_route: &[u8]) -> Option<Route> {
let version = version_of(fixed[0])?;
let destination_length = fixed[1];
let mut route = Route {
// Without a destination attribute: the default route.
prefix: IpPrefix::new(version, [0u8; 16], destination_length).ok()?,
table: fixed[4].into(),
protocol: fixed[5],
metric: 0,
gateway: None,
interface: None,
kind: fixed[7],
};
for (kind, value) in attributes(attributes_of_route) {
match kind {
route_attribute::DESTINATION => {
// The address in full, whatever the prefix's length.
address(version, value)?;
route.prefix = IpPrefix::new(version, value, destination_length).ok()?;
}
route_attribute::GATEWAY => route.gateway = Some(address(version, value)?),
route_attribute::OUT_INTERFACE => route.interface = Some(ne_u32(value.get(..4)?)),
route_attribute::PRIORITY => route.metric = ne_u32(value.get(..4)?),
route_attribute::TABLE => route.table = ne_u32(value.get(..4)?),
_ => {}
}
}
Some(route)
}
fn interface_address(fixed: &[u8], attributes_of_address: &[u8]) -> Option<Address> {
let version = version_of(fixed[0])?;
let prefix_length = fixed[1];
let (mut local, mut peer) = (None, None);
for (kind, value) in attributes(attributes_of_address) {
match kind {
address_attribute::LOCAL => local = Some(address(version, value)?),
address_attribute::ADDRESS => peer = Some(address(version, value)?),
_ => {}
}
}
// On an ordinary interface both are the interface's address, and IPv6
// gives only the one. On a point-to-point link the one is our end and
// the other the far end, which is what the prefix is of.
let local = local.or(peer)?;
let of_network = peer.unwrap_or(local);
let network = match of_network {
IpAddr::V4(v4) => IpPrefix::new(version, v4.octets(), prefix_length),
IpAddr::V6(v6) => IpPrefix::new(version, v6.octets(), prefix_length),
}
.ok()?;
Some(Address {
interface: ne_u32(&fixed[4..]),
local,
network,
})
}
/// What to do with a route.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Change {
/// Put it there, in place of the route there is for the same prefix
/// in the same table with the same metric, if there is one.
Replace,
Remove,
}
/// What to ask for all of.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Dump {
Routes(IpVersion),
Addresses,
Links,
}
/// Writes a header and returns where the message began, for
/// [`finish`] to fill in the length.
fn begin(out: &mut BytesMut, message_type: u16, flags: u16, sequence: u32) -> usize {
let start = out.len();
out.put_u32_ne(0);
out.put_u16_ne(message_type);
out.put_u16_ne(flags | flag::REQUEST);
out.put_u32_ne(sequence);
// From us: the kernel fills in who that is.
out.put_u32_ne(0);
start
}
fn finish(out: &mut BytesMut, start: usize) {
let length = (out.len() - start) as u32;
out[start..start + 4].copy_from_slice(&length.to_ne_bytes());
}
fn put_attribute(out: &mut BytesMut, kind: u16, value: &[u8]) {
out.put_u16_ne((4 + value.len()) as u16);
out.put_u16_ne(kind);
out.put_slice(value);
out.put_bytes(0, aligned(value.len()) - value.len());
}
fn octets(address: &IpAddr) -> ([u8; 16], usize) {
let mut bytes = [0u8; 16];
match address {
IpAddr::V4(v4) => {
bytes[..4].copy_from_slice(&v4.octets());
(bytes, 4)
}
IpAddr::V6(v6) => (v6.octets(), 16),
}
}
/// Writes the request to make `change` to `route` to the end of `out`.
/// With `acknowledge` the kernel answers also when it has done it; it
/// always answers when it could not.
pub fn route_request(
out: &mut BytesMut,
sequence: u32,
change: Change,
acknowledge: bool,
route: &Route,
) {
let (message_type, mut flags) = match change {
Change::Replace => (message_type::NEW_ROUTE, flag::CREATE | flag::REPLACE),
Change::Remove => (message_type::DEL_ROUTE, 0),
};
if acknowledge {
flags |= flag::ACK;
}
let start = begin(out, message_type, flags, sequence);
let version = route.prefix.address_family();
out.put_u8(family_of(version));
out.put_u8(route.prefix.length());
// No source prefix, no TOS.
out.put_u8(0);
out.put_u8(0);
out.put_u8(u8::try_from(route.table).unwrap_or(TABLE_IN_ATTRIBUTE));
out.put_u8(route.protocol);
out.put_u8(SCOPE_UNIVERSE);
out.put_u8(route.kind);
out.put_u32_ne(0);
let mut destination = [0u8; 16];
destination[..route.prefix.bytes().len()].copy_from_slice(route.prefix.bytes());
let address_length = usize::from(version.bits() / 8);
put_attribute(
out,
route_attribute::DESTINATION,
&destination[..address_length],
);
put_attribute(out, route_attribute::TABLE, &route.table.to_ne_bytes());
put_attribute(out, route_attribute::PRIORITY, &route.metric.to_ne_bytes());
if let Some(gateway) = &route.gateway {
let (bytes, length) = octets(gateway);
put_attribute(out, route_attribute::GATEWAY, &bytes[..length]);
}
if let Some(interface) = route.interface {
put_attribute(
out,
route_attribute::OUT_INTERFACE,
&interface.to_ne_bytes(),
);
}
finish(out, start);
}
/// Writes the request for all of `what` to the end of `out`. The answer is
/// a message for each, and then [`Message::Done`].
pub fn dump_request(out: &mut BytesMut, sequence: u32, what: Dump) {
let (message_type, family, fixed_length) = match what {
Dump::Routes(version) => (message_type::GET_ROUTE, family_of(version), 12),
Dump::Addresses => (message_type::GET_ADDRESS, 0, 8),
Dump::Links => (message_type::GET_LINK, 0, 16),
};
let start = begin(out, message_type, flag::DUMP, sequence);
// The fixed part of the kind asked for, empty but for the family.
out.put_u8(family);
out.put_bytes(0, fixed_length - 1);
finish(out, start);
}
#[cfg(test)]
mod tests {
use proptest::prelude::*;
use super::*;
fn prefix(s: &str) -> IpPrefix {
IpPrefix::try_from(s).unwrap()
}
/// What the kernel sends back for a request is the request's fixed part
/// and attributes under another type: a route that is written reads
/// back as itself.
fn read_back(route: &Route, change: Change) -> (Header, Message) {
let mut wire = BytesMut::new();
route_request(&mut wire, 7, change, true, route);
let (header, message, used) = decode(&wire).unwrap();
assert_eq!(used, wire.len());
(header, message)
}
#[test]
fn a_route_request_is_laid_out_as_the_kernel_expects() {
let route = Route {
prefix: prefix("203.0.113.0/24"),
table: TABLE_MAIN,
protocol: PROTOCOL_BGP,
metric: 20,
gateway: Some("192.0.2.1".parse().unwrap()),
interface: Some(3),
kind: KIND_UNICAST,
};
let mut wire = BytesMut::new();
route_request(&mut wire, 7, Change::Replace, true, &route);
let mut expected = Vec::new();
let ne16 = |v: u16| v.to_ne_bytes();
let ne32 = |v: u32| v.to_ne_bytes();
// nlmsghdr: length, RTM_NEWROUTE, REQUEST|ACK|REPLACE|CREATE,
// sequence, port.
expected.extend(ne32(16 + 12 + 5 * 8));
expected.extend(ne16(24));
expected.extend(ne16(0x1 | 0x4 | 0x100 | 0x400));
expected.extend(ne32(7));
expected.extend(ne32(0));
// rtmsg: AF_INET, /24, no source, no TOS, table main, proto BGP,
// scope universe, unicast, no flags.
expected.extend([2, 24, 0, 0, 254, 186, 0, 1]);
expected.extend(ne32(0));
// RTA_DST, RTA_TABLE, RTA_PRIORITY, RTA_GATEWAY, RTA_OIF.
for (kind, value) in [
(1u16, [203, 0, 113, 0]),
(15, ne32(254)),
(6, ne32(20)),
(5, [192, 0, 2, 1]),
(4, ne32(3)),
] {
expected.extend(ne16(8));
expected.extend(ne16(kind));
expected.extend(value);
}
assert_eq!(wire[..], expected[..]);
assert_eq!(
read_back(&route, Change::Replace),
(
Header {
flags: 0x505,
sequence: 7
},
Message::Route {
present: true,
route
}
)
);
assert_eq!(
read_back(&route, Change::Remove).1,
Message::Route {
present: false,
route
}
);
}
#[test]
fn an_ipv6_route_in_a_high_table_reads_back() {
let route = Route {
prefix: prefix("2001:db8:1::/48"),
// Does not fit the octet in the fixed part.
table: 10_000,
protocol: PROTOCOL_BGP,
metric: 20,
gateway: Some("fe80::1".parse().unwrap()),
interface: Some(9),
kind: KIND_UNICAST,
};
let mut wire = BytesMut::new();
route_request(&mut wire, 1, Change::Replace, false, &route);
assert_eq!(wire[16], 10, "AF_INET6");
assert_eq!(wire[20], 252, "the table is in the attribute");
assert_eq!(
read_back(&route, Change::Replace).1,
Message::Route {
present: true,
route
}
);
}
#[test]
fn dump_requests_carry_the_fixed_part_of_what_they_ask_for() {
for (what, message_type, length, family) in [
(Dump::Routes(IpVersion::Ipv4), 26u16, 28u32, 2u8),
(Dump::Routes(IpVersion::Ipv6), 26, 28, 10),
(Dump::Addresses, 22, 24, 0),
(Dump::Links, 18, 32, 0),
] {
let mut wire = BytesMut::new();
dump_request(&mut wire, 3, what);
assert_eq!(wire.len(), length as usize, "{what:?}");
assert_eq!(wire[..4], length.to_ne_bytes(), "{what:?}");
assert_eq!(wire[4..6], message_type.to_ne_bytes(), "{what:?}");
assert_eq!(wire[6..8], 0x301u16.to_ne_bytes(), "{what:?}");
assert_eq!(wire[16], family, "{what:?}");
}
}
/// A message as the kernel would send it: header, fixed part,
/// attributes.
fn message(message_type: u16, sequence: u32, fixed: &[u8], attrs: &[(u16, &[u8])]) -> Vec<u8> {
let mut out = BytesMut::new();
let start = begin(&mut out, message_type, 0, sequence);
out.put_slice(fixed);
for (kind, value) in attrs {
put_attribute(&mut out, *kind, value);
}
finish(&mut out, start);
out.to_vec()
}
#[test]
fn outcomes_addresses_and_links_are_read() {
// An error: ENETUNREACH is 101, negated; then the request's header.
let mut body = (-101i32).to_ne_bytes().to_vec();
body.extend([0u8; 16]);
let wire = message(2, 42, &body, &[]);
let (header, outcome, _) = decode(&wire).unwrap();
assert_eq!(header.sequence, 42);
assert_eq!(outcome, Message::Outcome { errno: 101 });
// An acknowledgement is an error of zero.
let wire = message(2, 43, &[0u8; 20], &[]);
assert_eq!(decode(&wire).unwrap().1, Message::Outcome { errno: 0 });
assert_eq!(
decode(&message(3, 1, &[0; 4], &[])).unwrap().1,
Message::Done
);
// 192.0.2.1/24 on interface 3: IFA_ADDRESS and IFA_LOCAL.
let mut fixed = vec![2u8, 24, 0, 0];
fixed.extend(3u32.to_ne_bytes());
let wire = message(20, 0, &fixed, &[(1, &[192, 0, 2, 1]), (2, &[192, 0, 2, 1])]);
assert_eq!(
decode(&wire).unwrap().1,
Message::Address {
present: true,
address: Address {
interface: 3,
local: "192.0.2.1".parse().unwrap(),
network: prefix("192.0.2.0/24"),
}
}
);
// A point-to-point link: our end, and the far end the prefix is
// of.
let mut fixed = vec![2u8, 32, 0, 0];
fixed.extend(4u32.to_ne_bytes());
let wire = message(21, 0, &fixed, &[(1, &[10, 0, 0, 2]), (2, &[10, 0, 0, 1])]);
assert_eq!(
decode(&wire).unwrap().1,
Message::Address {
present: false,
address: Address {
interface: 4,
local: "10.0.0.1".parse().unwrap(),
network: prefix("10.0.0.2/32"),
}
}
);
// IPv6 gives only IFA_ADDRESS.
let v6: Ipv6Addr = "2001:db8::1".parse().unwrap();
let mut fixed = vec![10u8, 64, 0, 0];
fixed.extend(5u32.to_ne_bytes());
let wire = message(20, 0, &fixed, &[(1, &v6.octets())]);
assert_eq!(
decode(&wire).unwrap().1,
Message::Address {
present: true,
address: Address {
interface: 5,
local: v6.into(),
network: prefix("2001:db8::/64"),
}
}
);
// A link that is up and running, and one that is only up.
for (flags, up) in [(0x41u32, true), (0x1, false), (0x40, false)] {
let mut fixed = vec![0u8, 0, 1, 0];
fixed.extend(7u32.to_ne_bytes());
fixed.extend(flags.to_ne_bytes());
fixed.extend(0u32.to_ne_bytes());
let wire = message(16, 0, &fixed, &[(3, b"eth0\0")]);
assert_eq!(
decode(&wire).unwrap().1,
Message::Link {
present: true,
link: Link {
index: 7,
name: "eth0".into(),
up
}
}
);
}
}
#[test]
fn several_messages_in_one_read_are_read_one_after_another() {
let mut wire = message(2, 1, &[0u8; 20], &[]);
// An attribute whose value is not a multiple of four pads the
// message.
let mut fixed = vec![0u8, 0, 1, 0];
fixed.extend(7u32.to_ne_bytes());
fixed.extend([0u8; 8]);
wire.extend(message(16, 0, &fixed, &[(3, b"lo\0")]));
wire.extend(message(3, 2, &[0; 4], &[]));
let mut rest = &wire[..];
let mut seen = Vec::new();
while !rest.is_empty() {
let (header, message, used) = decode(rest).unwrap();
seen.push((header.sequence, std::mem::discriminant(&message)));
rest = &rest[used..];
}
assert_eq!(seen.len(), 3);
assert_eq!(seen[0].0, 1);
assert_eq!(seen[2].0, 2);
}
#[test]
fn what_is_not_a_message_is_an_error() {
assert_eq!(
decode(&[0; 8]),
Err(NetlinkError::Length {
claimed: 16,
available: 8
})
);
// A length past the end, and one shorter than a header.
for claimed in [64u32, 8] {
let mut wire = message(3, 1, &[], &[]);
wire[..4].copy_from_slice(&claimed.to_ne_bytes());
assert_eq!(
decode(&wire),
Err(NetlinkError::Length {
claimed: claimed as usize,
available: 16
})
);
}
// A route with half a fixed part.
assert_eq!(
decode(&message(24, 1, &[2, 24, 0, 0], &[])),
Err(NetlinkError::Truncated { message_type: 24 })
);
// Of a family that is not IP: nothing of ours.
let mut fixed = vec![28u8, 20, 0, 0, 254, 0, 0, 1];
fixed.extend([0u8; 4]);
assert_eq!(
decode(&message(24, 1, &fixed, &[])).unwrap().1,
Message::Other { message_type: 24 }
);
}
fn any_route() -> impl Strategy<Value = Route> {
(
any::<bool>(),
any::<[u8; 16]>(),
0u8..=128,
any::<u32>(),
any::<u8>(),
any::<u32>(),
proptest::option::of(any::<[u8; 16]>()),
proptest::option::of(any::<u32>()),
)
.prop_map(
|(v6, bytes, length, table, protocol, metric, gateway, interface)| {
let version = if v6 { IpVersion::Ipv6 } else { IpVersion::Ipv4 };
let gateway = gateway.map(|bytes| match version {
IpVersion::Ipv4 => IpAddr::from(<[u8; 4]>::try_from(&bytes[..4]).unwrap()),
IpVersion::Ipv6 => IpAddr::from(bytes),
});
Route {
prefix: IpPrefix::new(version, bytes, length.min(version.bits())).unwrap(),
table,
protocol,
metric,
gateway,
interface,
kind: KIND_UNICAST,
}
},
)
}
proptest! {
#[test]
fn a_route_that_is_written_reads_back_equal(route in any_route(), remove: bool) {
let change = if remove { Change::Remove } else { Change::Replace };
let read = read_back(&route, change).1;
prop_assert_eq!(read, Message::Route { present: !remove, route });
}
/// Whatever arrives, decoding returns: a message or an error. What
/// it reads is within what arrived, and it always moves on.
#[test]
fn noise_is_never_more_than_an_error(bytes in proptest::collection::vec(any::<u8>(), 0..120)) {
if let Ok((_, _, used)) = decode(&bytes) {
prop_assert!(used > 0 && used <= bytes.len());
}
}
/// Noise behind a header that could be a message's, to get past it.
#[test]
fn noise_behind_a_plausible_header_is_never_more_than_an_error(
message_type in 0u16..30,
body in proptest::collection::vec(any::<u8>(), 0..100),
) {
let wire = message(message_type, 1, &body, &[]);
if let Ok((_, _, used)) = decode(&wire) {
prop_assert!(used > 0 && used <= wire.len());
}
}
}
}
+619
View File
@@ -0,0 +1,619 @@
//! What should be in the kernel's table, what is, and what to do about the
//! difference. No I/O here: this is told what the routing table's best
//! routes are and what the kernel said, and answers with the changes to
//! ask the kernel for.
use std::collections::{HashMap, HashSet};
use std::net::{IpAddr, Ipv6Addr};
use bgp_packet::message::NlriNextHop;
use bgp_primitives::constants::IpVersion;
use bgp_primitives::ip_prefix::IpPrefix;
use crate::interfaces::Interfaces;
/// Where the kernel is to send packets for a prefix.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct Target {
pub gateway: IpAddr,
/// The index of the interface the gateway is on.
pub interface: u32,
}
/// Why a route's next hop is not something the kernel can be given.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Unresolved {
/// The route has no next hop.
Missing,
/// The next hop is of the other IP version than the prefix.
OtherFamily,
/// The next hop is one of this machine's own addresses.
Ourselves,
/// The next hop is not in the network of any interface that is up.
/// Finding the way to such a next hop through other routes is not
/// done.
NotConnected,
}
impl Unresolved {
pub const ALL: [Self; 4] = [
Self::Missing,
Self::OtherFamily,
Self::Ourselves,
Self::NotConnected,
];
pub fn as_str(self) -> &'static str {
match self {
Self::Missing => "no next hop",
Self::OtherFamily => "next hop of the other IP version",
Self::Ourselves => "next hop is this machine",
Self::NotConnected => "next hop not directly connected",
}
}
}
fn is_link_local(address: &Ipv6Addr) -> bool {
address.segments()[0] & 0xffc0 == 0xfe80
}
/// Works out where the kernel is to send packets for a route to `prefix`
/// with the next hop `next_hop`, learned from the peer at `peer`.
///
/// Only a next hop that is a neighbor on one of our interfaces is
/// resolved. A link-local next hop means something only on one link, and
/// that is taken to be the link the peer itself is on.
pub fn resolve(
interfaces: &Interfaces,
prefix: &IpPrefix,
next_hop: Option<&NlriNextHop>,
peer: Option<IpAddr>,
) -> Result<Target, Unresolved> {
let next_hop = next_hop.ok_or(Unresolved::Missing)?;
let version = match next_hop {
NlriNextHop::Ipv4(_) => IpVersion::Ipv4,
NlriNextHop::Ipv6(_) | NlriNextHop::Ipv6WithLl { .. } => IpVersion::Ipv6,
};
if version != prefix.address_family() {
return Err(Unresolved::OtherFamily);
}
let direct = |gateway: IpAddr| {
if interfaces.is_local(gateway) {
return Err(Unresolved::Ourselves);
}
interfaces
.connected(gateway)
.map(|connected| Target {
gateway,
interface: connected.interface,
})
.ok_or(Unresolved::NotConnected)
};
// The interface the peer is a neighbor on, for a link-local gateway.
let on_peers_link = |gateway: Ipv6Addr| {
peer.and_then(|peer| interfaces.connected(peer))
.map(|connected| Target {
gateway: gateway.into(),
interface: connected.interface,
})
.ok_or(Unresolved::NotConnected)
};
match *next_hop {
NlriNextHop::Ipv4(gateway) => direct(gateway.into()),
NlriNextHop::Ipv6(gateway) if is_link_local(&gateway) => on_peers_link(gateway),
NlriNextHop::Ipv6(gateway) => direct(gateway.into()),
// The global address if it is a neighbor; else the link-local one,
// which is there for exactly that case (RFC 2545 §3).
NlriNextHop::Ipv6WithLl { global, link_local } => match direct(global.into()) {
Err(Unresolved::NotConnected) => on_peers_link(link_local),
resolved => resolved,
},
}
}
/// A change to ask the kernel for.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Op {
/// Put this route there, in place of ours for the prefix if there is
/// one.
Install {
prefix: IpPrefix,
target: Target,
},
Remove {
prefix: IpPrefix,
},
}
impl Op {
pub fn prefix(&self) -> &IpPrefix {
match self {
Self::Install { prefix, .. } | Self::Remove { prefix } => prefix,
}
}
}
struct Entry {
/// What the kernel has; `None` for a route found there at startup that
/// is not of a shape we make.
target: Option<Target>,
/// Whether the prefix was heard of since the last reset.
confirmed: bool,
}
/// The routes of ours in the kernel, kept in step with the best routes.
///
/// It follows what a reader of the routing table is given: the whole
/// table between a reset and its completion, and changes after. Whatever
/// is in the kernel and was not confirmed by the end of a whole table is
/// removed: that is how routes left by an earlier run, or made stale by a
/// change of interfaces, are cleared out.
#[derive(Default)]
pub struct Plan {
/// What the kernel has been asked to have, and has not refused.
installed: HashMap<IpPrefix, Entry>,
/// Prefixes whose best route is not installed, and why.
unresolved: HashMap<IpPrefix, Unresolved>,
/// How many of them for each reason, in the order of
/// [`Unresolved::ALL`].
unresolved_counts: [usize; 4],
/// Prefixes whose route the kernel refused.
refused: HashSet<IpPrefix>,
}
impl Plan {
/// How many routes of ours the kernel has.
pub fn installed(&self) -> usize {
self.installed.len()
}
/// How many prefixes have a best route that could not be given to the
/// kernel for want of a usable next hop.
pub fn unresolved(&self) -> usize {
self.unresolved.len()
}
/// The same, by reason, leaving out the reasons there are none of.
pub fn unresolved_by_reason(&self) -> impl Iterator<Item = (Unresolved, usize)> {
Unresolved::ALL
.into_iter()
.zip(self.unresolved_counts)
.filter(|(_, count)| *count > 0)
}
/// Notes why the best route of `prefix` is not installed, or with
/// `None` that this is no longer so.
fn set_unresolved(&mut self, prefix: IpPrefix, reason: Option<Unresolved>) {
let before = match reason {
Some(reason) => self.unresolved.insert(prefix, reason),
None => self.unresolved.remove(&prefix),
};
for (reason, by) in [(before, -1isize), (reason, 1)] {
if let Some(reason) = reason {
let count = &mut self.unresolved_counts[reason as usize];
*count = count.wrapping_add_signed(by);
}
}
}
/// How many prefixes have a best route that the kernel refused.
pub fn refused(&self) -> usize {
self.refused.len()
}
/// The prefixes of the routes the kernel has.
pub fn installed_prefixes(&self) -> impl Iterator<Item = &IpPrefix> {
self.installed.keys()
}
/// Notes a route of ours that the kernel already has, from before this
/// run. It stays only if the routing table turns out to have the same.
pub fn adopt(&mut self, prefix: IpPrefix, target: Option<Target>) {
self.installed.insert(
prefix,
Entry {
target,
confirmed: false,
},
);
}
/// The whole table follows: nothing is known to be wanted until it is
/// heard of again.
pub fn reset(&mut self) {
for entry in self.installed.values_mut() {
entry.confirmed = false;
}
self.unresolved.clear();
self.unresolved_counts = [0; 4];
self.refused.clear();
}
/// The best route for `prefix` is now one that resolves as `target`.
pub fn best(&mut self, prefix: IpPrefix, target: Result<Target, Unresolved>) -> Option<Op> {
self.refused.remove(&prefix);
match target {
Ok(target) => {
self.set_unresolved(prefix, None);
let entry = self.installed.entry(prefix).or_insert(Entry {
target: None,
confirmed: true,
});
entry.confirmed = true;
if entry.target == Some(target) {
return None;
}
entry.target = Some(target);
Some(Op::Install { prefix, target })
}
Err(reason) => {
self.set_unresolved(prefix, Some(reason));
// Packets must not go on following a route that is no
// longer the best.
self.installed
.remove(&prefix)
.map(|_| Op::Remove { prefix })
}
}
}
/// `prefix` has no route any more.
pub fn withdrawn(&mut self, prefix: IpPrefix) -> Option<Op> {
self.set_unresolved(prefix, None);
self.refused.remove(&prefix);
self.installed
.remove(&prefix)
.map(|_| Op::Remove { prefix })
}
/// The whole table has been given: what was not in it goes.
pub fn synced(&mut self) -> Vec<Op> {
let stale: Vec<IpPrefix> = self
.installed
.iter()
.filter(|(_, entry)| !entry.confirmed)
.map(|(prefix, _)| *prefix)
.collect();
stale
.into_iter()
.map(|prefix| {
self.installed.remove(&prefix);
Op::Remove { prefix }
})
.collect()
}
/// The kernel refused `op`. Returns what to ask of it instead, if
/// anything.
pub fn refused_by_kernel(&mut self, op: &Op) -> Option<Op> {
match op {
Op::Install { prefix, target } => {
// If the plan has moved on since, this is old news.
if self.installed.get(prefix)?.target != Some(*target) {
return None;
}
self.installed.remove(prefix);
self.refused.insert(*prefix);
// What the kernel refused to replace, it still has.
Some(Op::Remove { prefix: *prefix })
}
// It was not there to remove: as wanted.
Op::Remove { .. } => None,
}
}
/// Everything goes: the changes that empty the kernel of our routes.
pub fn clear(&mut self) -> Vec<Op> {
self.unresolved.clear();
self.unresolved_counts = [0; 4];
self.refused.clear();
self.installed
.drain()
.map(|(prefix, _)| Op::Remove { prefix })
.collect()
}
}
#[cfg(test)]
mod tests {
use proptest::prelude::*;
use super::*;
use crate::netlink::{Address, Link, Message};
fn prefix(s: &str) -> IpPrefix {
IpPrefix::try_from(s).unwrap()
}
fn target(gateway: &str, interface: u32) -> Target {
Target {
gateway: gateway.parse().unwrap(),
interface,
}
}
fn interfaces() -> Interfaces {
let mut interfaces = Interfaces::default();
for (index, local, network) in [
(2, "192.0.2.1", "192.0.2.0/24"),
(2, "2001:db8::1", "2001:db8::/64"),
(3, "198.51.100.1", "198.51.100.0/24"),
] {
interfaces.apply(&Message::Link {
present: true,
link: Link {
index,
name: format!("eth{index}"),
up: true,
},
});
interfaces.apply(&Message::Address {
present: true,
address: Address {
interface: index,
local: local.parse().unwrap(),
network: prefix(network),
},
});
}
interfaces
}
#[test]
fn a_next_hop_resolves_if_it_is_a_neighbor() {
let interfaces = interfaces();
let v4 = prefix("203.0.113.0/24");
let v6 = prefix("2001:db8:100::/48");
let peer = Some("192.0.2.2".parse().unwrap());
let v4_hop = |s: &str| NlriNextHop::Ipv4(s.parse().unwrap());
let v6_hop = |s: &str| NlriNextHop::Ipv6(s.parse().unwrap());
assert_eq!(
resolve(&interfaces, &v4, Some(&v4_hop("192.0.2.2")), peer),
Ok(target("192.0.2.2", 2))
);
// A neighbor other than the peer: a route server's routes.
assert_eq!(
resolve(&interfaces, &v4, Some(&v4_hop("198.51.100.9")), peer),
Ok(target("198.51.100.9", 3))
);
assert_eq!(
resolve(&interfaces, &v6, Some(&v6_hop("2001:db8::2")), peer),
Ok(target("2001:db8::2", 2))
);
for (hop, why) in [
(Some(v4_hop("10.9.9.9")), Unresolved::NotConnected),
(Some(v4_hop("192.0.2.1")), Unresolved::Ourselves),
(Some(v6_hop("2001:db8::2")), Unresolved::OtherFamily),
(None, Unresolved::Missing),
] {
assert_eq!(resolve(&interfaces, &v4, hop.as_ref(), peer), Err(why));
}
assert_eq!(
resolve(&interfaces, &v6, Some(&v4_hop("192.0.2.2")), peer),
Err(Unresolved::OtherFamily)
);
}
#[test]
fn a_link_local_next_hop_is_on_the_peers_link() {
let interfaces = interfaces();
let v6 = prefix("2001:db8:100::/48");
let peer = Some("2001:db8::2".parse().unwrap());
let link_local: Ipv6Addr = "fe80::2".parse().unwrap();
let both = |global: &str| NlriNextHop::Ipv6WithLl {
global: global.parse().unwrap(),
link_local,
};
// The global address is a neighbor: it is used.
assert_eq!(
resolve(&interfaces, &v6, Some(&both("2001:db8::2")), peer),
Ok(target("2001:db8::2", 2))
);
// It is not: the link-local one, on the interface the peer is on.
assert_eq!(
resolve(&interfaces, &v6, Some(&both("2001:db8:ffff::2")), peer),
Ok(target("fe80::2", 2))
);
assert_eq!(
resolve(&interfaces, &v6, Some(&NlriNextHop::Ipv6(link_local)), peer),
Ok(target("fe80::2", 2))
);
// A peer that is not a neighbor says nothing about the link.
let far = Some("2001:db8:ffff::9".parse().unwrap());
assert_eq!(
resolve(&interfaces, &v6, Some(&NlriNextHop::Ipv6(link_local)), far),
Err(Unresolved::NotConnected)
);
assert_eq!(
resolve(&interfaces, &v6, Some(&both("2001:db8::1")), peer),
Err(Unresolved::Ourselves)
);
}
#[test]
fn the_kernel_is_asked_only_for_what_changes() {
let mut plan = Plan::default();
let p = prefix("203.0.113.0/24");
let (a, b) = (target("192.0.2.2", 2), target("192.0.2.3", 2));
assert_eq!(
plan.best(p, Ok(a)),
Some(Op::Install {
prefix: p,
target: a
})
);
// The best route changes to one with the same next hop: nothing
// for the kernel.
assert_eq!(plan.best(p, Ok(a)), None);
assert_eq!(
plan.best(p, Ok(b)),
Some(Op::Install {
prefix: p,
target: b
})
);
assert_eq!(plan.installed(), 1);
// A best route that cannot be installed takes the old one out.
assert_eq!(
plan.best(p, Err(Unresolved::NotConnected)),
Some(Op::Remove { prefix: p })
);
assert_eq!((plan.installed(), plan.unresolved()), (0, 1));
assert_eq!(
plan.unresolved_by_reason().collect::<Vec<_>>(),
[(Unresolved::NotConnected, 1)]
);
assert_eq!(plan.withdrawn(p), None);
assert_eq!(plan.unresolved_by_reason().count(), 0);
assert_eq!(plan.unresolved(), 0);
plan.best(p, Ok(a));
assert_eq!(plan.withdrawn(p), Some(Op::Remove { prefix: p }));
assert_eq!(plan.withdrawn(p), None);
}
#[test]
fn what_a_whole_table_does_not_confirm_is_removed() {
let mut plan = Plan::default();
let (kept, changed, stale, odd) = (
prefix("203.0.113.0/24"),
prefix("198.18.0.0/15"),
prefix("192.88.99.0/24"),
prefix("2001:db8:5::/48"),
);
let a = target("192.0.2.2", 2);
// Left in the kernel by an earlier run.
plan.adopt(kept, Some(a));
plan.adopt(changed, Some(a));
plan.adopt(stale, Some(a));
plan.adopt(odd, None);
assert_eq!(plan.installed(), 4);
plan.reset();
// The same as the kernel has: nothing to do.
assert_eq!(plan.best(kept, Ok(a)), None);
let b = target("192.0.2.3", 2);
assert_eq!(
plan.best(changed, Ok(b)),
Some(Op::Install {
prefix: changed,
target: b
})
);
let mut removed = plan.synced();
removed.sort_by_key(|op| *op.prefix());
let mut expected = vec![Op::Remove { prefix: stale }, Op::Remove { prefix: odd }];
expected.sort_by_key(|op| *op.prefix());
assert_eq!(removed, expected);
assert_eq!(plan.installed(), 2);
// A second completion finds nothing more.
assert_eq!(plan.synced(), []);
}
#[test]
fn a_route_the_kernel_refuses_is_not_counted_as_there() {
let mut plan = Plan::default();
let p = prefix("203.0.113.0/24");
let (a, b) = (target("192.0.2.2", 2), target("192.0.2.3", 2));
let install = plan.best(p, Ok(a)).unwrap();
// What it refused to put there may have left an older one of ours.
assert_eq!(
plan.refused_by_kernel(&install),
Some(Op::Remove { prefix: p })
);
assert_eq!((plan.installed(), plan.refused()), (0, 1));
// Refusing to remove what is not there is as wanted.
assert_eq!(plan.refused_by_kernel(&Op::Remove { prefix: p }), None);
// The next best route is tried afresh.
assert!(plan.best(p, Ok(b)).is_some());
assert_eq!((plan.installed(), plan.refused()), (1, 0));
// News of a refusal of what has since been replaced is ignored.
assert_eq!(plan.refused_by_kernel(&install), None);
assert_eq!(plan.installed(), 1);
let mut cleared = plan.clear();
assert_eq!(cleared.pop(), Some(Op::Remove { prefix: p }));
assert_eq!(plan.installed(), 0);
}
/// What happens to a table of best routes.
#[derive(Debug, Clone)]
enum Step {
Best(u8, Option<u8>),
Withdrawn(u8),
/// A reader that starts over: the whole table again.
Resync,
}
fn any_step() -> impl Strategy<Value = Step> {
prop_oneof![
4 => (0u8..12, proptest::option::weighted(0.8, 0u8..3)).prop_map(|(p, t)| Step::Best(p, t)),
2 => (0u8..12).prop_map(Step::Withdrawn),
1 => Just(Step::Resync),
]
}
proptest! {
/// Whatever happens to the best routes, and whatever an earlier
/// run left behind, a kernel that does as it is asked ends up with
/// exactly the best routes that resolve.
#[test]
fn the_kernel_ends_up_with_the_best_routes_that_resolve(
left_behind in proptest::collection::vec((0u8..16, 0u8..3), 0..6),
steps in proptest::collection::vec(any_step(), 0..60),
) {
let p = |n: u8| IpPrefix::new(IpVersion::Ipv4, [10, n, 0, 0], 16).unwrap();
let t = |n: u8| target(&format!("192.0.2.{}", n + 2), 2);
let mut kernel: HashMap<IpPrefix, Target> = HashMap::new();
let mut plan = Plan::default();
for (prefix, gateway) in left_behind {
kernel.insert(p(prefix), t(gateway));
plan.adopt(p(prefix), Some(t(gateway)));
}
// The best routes: the prefix's next hop, if it resolves.
let mut table: HashMap<u8, Option<u8>> = HashMap::new();
let apply = |kernel: &mut HashMap<IpPrefix, Target>, op: Option<Op>| match op {
Some(Op::Install { prefix, target }) => { kernel.insert(prefix, target); }
Some(Op::Remove { prefix }) => { kernel.remove(&prefix); }
None => {}
};
let resolved = |gateway: Option<u8>| gateway.map(t).ok_or(Unresolved::NotConnected);
// A new reader is first given the whole table.
let steps = std::iter::once(Step::Resync).chain(steps);
for step in steps {
match step {
Step::Best(prefix, gateway) => {
table.insert(prefix, gateway);
apply(&mut kernel, plan.best(p(prefix), resolved(gateway)));
}
Step::Withdrawn(prefix) => {
table.remove(&prefix);
apply(&mut kernel, plan.withdrawn(p(prefix)));
}
Step::Resync => {
plan.reset();
for (prefix, gateway) in &table {
apply(&mut kernel, plan.best(p(*prefix), resolved(*gateway)));
}
for op in plan.synced() {
apply(&mut kernel, Some(op));
}
}
}
let wanted: HashMap<IpPrefix, Target> = table
.iter()
.filter_map(|(prefix, gateway)| Some((p(*prefix), t((*gateway)?))))
.collect();
prop_assert_eq!(&kernel, &wanted);
prop_assert_eq!(plan.installed(), wanted.len());
prop_assert_eq!(plan.unresolved(), table.len() - wanted.len());
prop_assert_eq!(
plan.unresolved_by_reason().map(|(_, count)| count).sum::<usize>(),
plan.unresolved()
);
}
}
}
}
+417
View File
@@ -0,0 +1,417 @@
//! The kernel's routes against a real kernel: what `ip route` shows after
//! the routing table changes.
//!
//! These change the machine's interfaces and routes, so they are not run
//! by `cargo test`. `scripts/kernel-tests.sh` runs them in a network
//! namespace of their own, with the capability they need
//! (CAP_NET_ADMIN) and nothing more.
#![cfg(target_os = "linux")]
use std::collections::BTreeSet;
use std::net::{IpAddr, Ipv4Addr};
use std::process::Command;
use std::sync::Arc;
use std::time::{Duration, Instant};
use bgp_fib::{Fib, FibConfig, FibError};
use bgp_packet::message::{
AsPathAttribute, MultiExitDiscPathAttribute, NextHopPathAttribute, NlriNextHop,
OriginPathAttribute, PathAttribute,
};
use bgp_primitives::ip_prefix::IpPrefix;
use bgp_rib::table::{AttrSet, PeerIndex, PeerInfo, RibTable};
use tokio::task::JoinHandle;
const TIMEOUT: Duration = Duration::from_secs(10);
fn prefix(s: &str) -> IpPrefix {
IpPrefix::try_from(s).unwrap()
}
/// Runs `ip` and returns what it printed. Panics if it fails.
#[track_caller]
fn ip(arguments: &str) -> String {
let output = Command::new("ip")
.args(arguments.split_whitespace())
.output()
.expect("the ip command is there");
assert!(
output.status.success(),
"ip {arguments}: {}",
String::from_utf8_lossy(&output.stderr)
);
String::from_utf8_lossy(&output.stdout).into_owned()
}
/// What a test has of the machine: an interface of its own with
/// 10.`n`.0.1/24 and 2001:db8:`n`::1/64 on it, and a routing table of its
/// own. Tests run side by side in one network namespace, and `n` keeps
/// them apart.
struct Machine {
n: u8,
rib: Arc<RibTable>,
}
impl Machine {
fn new(n: u8) -> Self {
let machine = Self {
n,
rib: Arc::new(RibTable::new(4)),
};
ip(&format!("link add {} type dummy", machine.interface()));
machine.add_addresses();
ip(&format!("link set {} up", machine.interface()));
machine
}
fn interface(&self) -> String {
format!("ferrix{}", self.n)
}
fn add_addresses(&self) {
let (n, interface) = (self.n, self.interface());
ip(&format!("addr add 10.{n}.0.1/24 dev {interface}"));
ip(&format!(
"-6 addr add 2001:db8:{n}::1/64 dev {interface} nodad"
));
}
fn table(&self) -> u32 {
1000 + u32::from(self.n)
}
/// A neighbor on the test's interface.
fn neighbor(&self, host: u8) -> Ipv4Addr {
Ipv4Addr::new(10, self.n, 0, host)
}
fn start(&self) -> (Fib, JoinHandle<Result<(), FibError>>) {
Fib::spawn(
Arc::clone(&self.rib),
FibConfig {
table: self.table(),
metric: 20,
},
)
.expect("the kernel can be talked to")
}
fn peer(&self, host: u8) -> PeerIndex {
self.rib
.add_peer(PeerInfo {
asn: 65000 + u32::from(host),
router_id: self.neighbor(host),
address: self.neighbor(host).into(),
ibgp: false,
})
.unwrap()
}
/// Attributes with the next hop `next_hop` and, to say which of two
/// routes is the better, `med`.
fn via(&self, next_hop: IpAddr, med: u32) -> Arc<AttrSet> {
let mut attributes = vec![
PathAttribute::Origin(OriginPathAttribute::IGP),
AsPathAttribute::from_asns(vec![65001]),
PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(med)),
];
let mp_next_hop = match next_hop {
IpAddr::V4(v4) => {
attributes.push(PathAttribute::NextHop(NextHopPathAttribute(v4)));
None
}
IpAddr::V6(v6) => Some(NlriNextHop::Ipv6(v6)),
};
self.rib
.attrs()
.intern(&attributes, mp_next_hop.as_ref())
.unwrap()
}
/// The routes in the test's table, as `ip route` prints them.
fn kernel_routes(&self) -> BTreeSet<String> {
let table = self.table();
let mut routes = BTreeSet::new();
for family in ["-4", "-6"] {
// A table nothing was ever put in does not exist yet.
let output = Command::new("ip")
.args([family, "route", "show", "table", &table.to_string()])
.output()
.unwrap();
routes.extend(
String::from_utf8_lossy(&output.stdout)
.lines()
.map(|line| line.split_whitespace().collect::<Vec<_>>().join(" "))
// IPv6 routes are shown with "pref medium".
.map(|line| line.replace(" pref medium", ""))
.filter(|line| !line.is_empty()),
);
}
routes
}
/// Waits until the kernel's table holds exactly `expected`.
#[track_caller]
fn kernel_becomes(&self, expected: &[String]) {
let expected: BTreeSet<String> = expected.iter().cloned().collect();
let start = Instant::now();
while self.kernel_routes() != expected {
assert!(
start.elapsed() < TIMEOUT,
"the kernel has {:#?}; expected {expected:#?}",
self.kernel_routes()
);
std::thread::sleep(Duration::from_millis(20));
}
}
/// How `ip route` shows a route of ours.
fn shown(&self, prefix: &str, gateway: impl std::fmt::Display) -> String {
format!(
"{prefix} via {gateway} dev {} proto bgp metric 20",
self.interface()
)
}
}
impl Drop for Machine {
fn drop(&mut self) {
let _ = Command::new("ip")
.args(["link", "del", &self.interface()])
.output();
for family in ["-4", "-6"] {
let _ = Command::new("ip")
.args([family, "route", "flush", "table", &self.table().to_string()])
.output();
}
}
}
/// Waits for `condition` to hold of the status.
#[track_caller]
fn status_becomes(fib: &Fib, what: &str, condition: impl Fn(&bgp_fib::Status) -> bool) {
let start = Instant::now();
while !condition(&fib.status()) {
assert!(
start.elapsed() < TIMEOUT,
"{what}: the status is {:#?}",
fib.status()
);
std::thread::sleep(Duration::from_millis(20));
}
}
/// The best route of each prefix is in the kernel, and follows the table:
/// when the best route changes, when it goes, and when there is none whose
/// next hop can be reached.
#[tokio::test(flavor = "multi_thread")]
#[ignore = "changes the machine's routes; run scripts/kernel-tests.sh"]
async fn the_kernel_has_the_best_route_of_each_prefix() {
let machine = Machine::new(1);
let (near, far) = (machine.peer(2), machine.peer(3));
let via_near = machine.via(machine.neighbor(2).into(), 50);
// What was there before the daemon started is installed too.
machine
.rib
.announce(near, &via_near, &[prefix("203.0.113.0/24")]);
let (fib, task) = machine.start();
machine.kernel_becomes(&[machine.shown("203.0.113.0/24", machine.neighbor(2))]);
status_becomes(&fib, "in step", |status| {
status.synced && status.installed == 1
});
// More routes, IPv6 among them.
let v6_neighbor: IpAddr = "2001:db8:1::2".parse().unwrap();
machine.rib.announce(
near,
&machine.via(v6_neighbor, 50),
&[prefix("2001:db8:100::/48")],
);
machine
.rib
.announce(near, &via_near, &[prefix("198.18.0.0/15")]);
machine.kernel_becomes(&[
machine.shown("203.0.113.0/24", machine.neighbor(2)),
machine.shown("198.18.0.0/15", machine.neighbor(2)),
machine.shown("2001:db8:100::/48", v6_neighbor),
]);
// A better route from another peer takes the first one's place.
let via_far = machine.via(machine.neighbor(3).into(), 10);
machine
.rib
.announce(far, &via_far, &[prefix("203.0.113.0/24")]);
machine.kernel_becomes(&[
machine.shown("203.0.113.0/24", machine.neighbor(3)),
machine.shown("198.18.0.0/15", machine.neighbor(2)),
machine.shown("2001:db8:100::/48", v6_neighbor),
]);
// It goes: the first is the best again.
machine.rib.withdraw(far, &[prefix("203.0.113.0/24")]);
machine.kernel_becomes(&[
machine.shown("203.0.113.0/24", machine.neighbor(2)),
machine.shown("198.18.0.0/15", machine.neighbor(2)),
machine.shown("2001:db8:100::/48", v6_neighbor),
]);
// A best route whose next hop is not a neighbor is not installed, and
// the route it replaced does not stay.
let unreachable = machine.via("172.31.9.9".parse().unwrap(), 5);
machine
.rib
.announce(far, &unreachable, &[prefix("198.18.0.0/15")]);
machine.kernel_becomes(&[
machine.shown("203.0.113.0/24", machine.neighbor(2)),
machine.shown("2001:db8:100::/48", v6_neighbor),
]);
status_becomes(&fib, "one unresolved", |status| {
status.installed == 2 && status.unresolved == [("next hop not directly connected", 1)]
});
// A session ends: its routes go.
machine.rib.remove_peer(near);
machine.kernel_becomes(&[]);
status_becomes(&fib, "nothing installed", |status| status.installed == 0);
fib.stop().await;
task.await.unwrap().unwrap();
}
/// Routes of ours left by a run that ended badly are removed at the next
/// start unless they are still wanted. Routes that are not ours are left
/// alone, whatever they are for.
#[tokio::test(flavor = "multi_thread")]
#[ignore = "changes the machine's routes; run scripts/kernel-tests.sh"]
async fn routes_left_behind_are_cleared_and_others_are_not_touched() {
let machine = Machine::new(2);
let (table, interface) = (machine.table(), machine.interface());
let add = |route: &str| ip(&format!("route add {route} dev {interface} table {table}"));
// Ours, from before: one that is still right, one that has another
// next hop now, one that nobody announces any more.
add("203.0.113.0/24 via 10.2.0.2 proto bgp metric 20");
add("198.18.0.0/15 via 10.2.0.9 proto bgp metric 20");
add("192.88.99.0/24 via 10.2.0.2 proto bgp metric 20");
// Not ours: someone's static route, and a BGP route with another
// metric, which is how another daemon's would differ.
add("192.0.2.0/24 via 10.2.0.7 proto static metric 20");
add("192.88.99.0/24 via 10.2.0.7 proto bgp metric 30");
let others = [
format!("192.0.2.0/24 via 10.2.0.7 dev {interface} proto static metric 20"),
format!("192.88.99.0/24 via 10.2.0.7 dev {interface} proto bgp metric 30"),
];
let peer = machine.peer(2);
let via = machine.via(machine.neighbor(2).into(), 0);
machine.rib.announce(
peer,
&via,
&[prefix("203.0.113.0/24"), prefix("198.18.0.0/15")],
);
let (fib, task) = machine.start();
machine.kernel_becomes(&[
machine.shown("203.0.113.0/24", machine.neighbor(2)),
machine.shown("198.18.0.0/15", machine.neighbor(2)),
others[0].clone(),
others[1].clone(),
]);
// A run that ends in good order takes its routes with it, and only
// its own.
fib.stop().await;
task.await.unwrap().unwrap();
machine.kernel_becomes(&others);
}
/// What a next hop resolves to depends on the machine's interfaces. When
/// they change, the routes are worked out again.
#[tokio::test(flavor = "multi_thread")]
#[ignore = "changes the machine's routes; run scripts/kernel-tests.sh"]
async fn routes_follow_the_machines_interfaces() {
let machine = Machine::new(3);
let interface = machine.interface();
// The interface has no addresses yet: nothing is a neighbor.
ip(&format!("addr flush dev {interface}"));
let peer = machine.peer(2);
let via = machine.via(machine.neighbor(2).into(), 0);
machine
.rib
.announce(peer, &via, &[prefix("203.0.113.0/24")]);
let (fib, task) = machine.start();
status_becomes(&fib, "unresolved", |status| {
status.synced && status.unresolved == [("next hop not directly connected", 1)]
});
assert!(machine.kernel_routes().is_empty());
// It gets its address: the next hop is a neighbor now.
machine.add_addresses();
machine.kernel_becomes(&[machine.shown("203.0.113.0/24", machine.neighbor(2))]);
status_becomes(&fib, "the interface is shown", |status| {
status.unresolved.is_empty()
&& status.interfaces.iter().any(|shown| {
shown.name == interface
&& shown.up
&& shown
.addresses
.contains(&(IpAddr::from(machine.neighbor(1)), 24))
})
});
// It loses it. The kernel drops the route by itself; what matters is
// that we know it has.
ip(&format!("addr del 10.3.0.1/24 dev {interface}"));
machine.kernel_becomes(&[]);
status_becomes(&fib, "unresolved again", |status| {
status.installed == 0 && status.unresolved.len() == 1
});
// And gets it back, with the interface having been down meanwhile.
ip(&format!("link set {interface} down"));
ip(&format!("addr add 10.3.0.1/24 dev {interface}"));
ip(&format!("link set {interface} up"));
machine.kernel_becomes(&[machine.shown("203.0.113.0/24", machine.neighbor(2))]);
fib.stop().await;
task.await.unwrap().unwrap();
machine.kernel_becomes(&[]);
}
/// A table's worth of routes goes in, and comes out, in reasonable time.
#[tokio::test(flavor = "multi_thread")]
#[ignore = "changes the machine's routes; run scripts/kernel-tests.sh"]
async fn many_routes_are_installed_and_removed() {
const ROUTES: usize = 200_000;
let machine = Machine::new(4);
let peer = machine.peer(2);
let via = machine.via(machine.neighbor(2).into(), 0);
let prefixes: Vec<IpPrefix> = (0..ROUTES as u32)
.map(|n| {
let [_, a, b, c] = n.to_be_bytes();
IpPrefix::new(
bgp_primitives::constants::IpVersion::Ipv4,
[100 + a, b, c, 0],
24,
)
.unwrap()
})
.collect();
machine.rib.announce(peer, &via, &prefixes);
let started = Instant::now();
let (fib, task) = machine.start();
status_becomes(&fib, "all installed", |status| {
status.synced && status.installed == ROUTES
});
let installed_in = started.elapsed();
assert_eq!(machine.kernel_routes().len(), ROUTES);
assert_eq!(fib.status().errors, 0, "{:?}", fib.status().last_error);
let started = Instant::now();
fib.stop().await;
task.await.unwrap().unwrap();
let removed_in = started.elapsed();
assert!(machine.kernel_routes().is_empty());
println!(
"{ROUTES} routes: installed in {installed_in:?} ({:.0}/s), removed in {removed_in:?}",
ROUTES as f64 / installed_in.as_secs_f64()
);
}
+1
View File
@@ -9,6 +9,7 @@ version.workspace = true
publish = false
[dependencies]
bgp-fib.workspace = true
bgp-packet.workspace = true
bgp-primitives.workspace = true
bgp-rib.workspace = true
+8 -3
View File
@@ -79,7 +79,7 @@ impl TestPeer {
match self.recv().await? {
Message::Update(u) => return Ok(u),
Message::KeepAlive => {}
other => bail!("Expected UPDATE, got {:?}", other),
other => return Err(eyre::eyre!("Expected UPDATE, got {:?}", other)),
}
}
}
@@ -96,7 +96,7 @@ impl TestPeer {
self.ctx.four_octet_asn = four_byte && self.offered_four_octet_asn;
Ok(o)
}
other => bail!("Expected OPEN, got {:?}", other),
other => Err(eyre::eyre!("Expected OPEN, got {:?}", other)),
}
}
@@ -189,7 +189,12 @@ impl TestPeer {
// Expect server KEEPALIVE
match self.recv().await? {
Message::KeepAlive => {}
other => bail!("Expected KEEPALIVE from server, got {:?}", other),
other => {
return Err(eyre::eyre!(
"Expected KEEPALIVE from server, got {:?}",
other
));
}
}
// Send our KEEPALIVE
self.send(&Message::KeepAlive).await?;
@@ -1,6 +1,7 @@
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;
use bgp_fib::Fib;
use bgp_packet::message::{NlriNextHop, PathAttribute};
use bgp_primitives::ip_prefix::IpPrefix;
use bgp_rib::LocalRib;
@@ -32,6 +33,8 @@ pub struct TestServer {
pub refusals: RefusalLog,
/// The data for route origin validation, if the server validates.
pub rpki: Option<Rpki>,
/// What keeps the best routes in the kernel, if the server does.
pub kernel: Option<Fib>,
pub local_asn: u32,
}
@@ -63,6 +66,7 @@ impl TestServer {
originate: vec![],
http_endpoint: None,
rpki: None,
kernel: None,
};
customize(&mut config);
@@ -77,6 +81,7 @@ impl TestServer {
let registry = bound.registry.clone();
let refusals = bound.refusals.clone();
let rpki = bound.rpki.clone();
let kernel = bound.kernel.clone();
let handle = tokio::spawn(bound.run());
Ok(Self {
@@ -87,6 +92,7 @@ impl TestServer {
registry,
refusals,
rpki,
kernel,
local_asn,
})
}
@@ -206,6 +212,7 @@ impl TestServer {
rib: Arc::clone(&self.rib),
refusals: self.refusals.clone(),
rpki: self.rpki.clone(),
kernel: self.kernel.clone(),
};
let handle = tokio::spawn(bgp_web::serve_on(state, listener));
Ok((base_url, HttpGuard { handle }))
+169
View File
@@ -0,0 +1,169 @@
//! The whole daemon against a real kernel: a peer announces routes over a
//! session, and `ip route` shows them.
//!
//! This changes the machine's interfaces and routes, so it is not run by
//! `cargo test`; `scripts/kernel-tests.sh` runs it in a network namespace
//! of its own.
#![cfg(target_os = "linux")]
use std::collections::BTreeSet;
use std::net::Ipv4Addr;
use std::process::Command;
use std::time::{Duration, Instant};
use bgp_fib::FibConfig;
use bgp_integration_tests::test_peer::TestPeer;
use bgp_integration_tests::test_server::TestServer;
use bgp_packet::message::{
AsPathAttribute, Message, NextHopPathAttribute, OriginPathAttribute, PathAttribute,
UpdateMessage,
};
use bgp_primitives::ip_prefix::IpPrefix;
use eyre::Result;
use serde_json::Value;
const TIMEOUT: Duration = Duration::from_secs(10);
const TABLE: u32 = 1100;
const INTERFACE: &str = "ferrix100";
#[track_caller]
fn ip(arguments: &str) {
let output = Command::new("ip")
.args(arguments.split_whitespace())
.output()
.expect("the ip command is there");
assert!(
output.status.success(),
"ip {arguments}: {}",
String::from_utf8_lossy(&output.stderr)
);
}
fn kernel_routes() -> BTreeSet<String> {
let output = Command::new("ip")
.args(["route", "show", "table", &TABLE.to_string()])
.output()
.unwrap();
String::from_utf8_lossy(&output.stdout)
.lines()
.map(|line| line.split_whitespace().collect::<Vec<_>>().join(" "))
.filter(|line| !line.is_empty())
.collect()
}
#[track_caller]
fn kernel_becomes(expected: &[&str]) {
let expected: BTreeSet<String> = expected.iter().map(|route| route.to_string()).collect();
let start = Instant::now();
while kernel_routes() != expected {
assert!(
start.elapsed() < TIMEOUT,
"the kernel has {:#?}; expected {expected:#?}",
kernel_routes()
);
std::thread::sleep(Duration::from_millis(20));
}
}
/// Removes the test's interface, and its routes with it, when dropped.
struct Interface;
impl Drop for Interface {
fn drop(&mut self) {
let _ = Command::new("ip").args(["link", "del", INTERFACE]).output();
let _ = Command::new("ip")
.args(["route", "flush", "table", &TABLE.to_string()])
.output();
}
}
async fn announce(peer: &mut TestPeer, prefixes: &[&str], next_hop: Ipv4Addr) -> Result<()> {
peer.send(&Message::Update(UpdateMessage {
withdrawn_routes: vec![],
path_attributes: vec![
PathAttribute::Origin(OriginPathAttribute::IGP),
AsPathAttribute::from_asns(vec![65001]),
PathAttribute::NextHop(NextHopPathAttribute(next_hop)),
],
announced_routes: prefixes
.iter()
.map(|p| IpPrefix::try_from(*p).unwrap().into())
.collect(),
}))
.await
}
/// A peer's routes go from the session to the kernel, and out of it again
/// when they are withdrawn and when the session ends.
#[tokio::test(flavor = "multi_thread")]
#[ignore = "changes the machine's routes; run scripts/kernel-tests.sh"]
async fn test_routes_from_a_session_reach_the_kernel() -> Result<()> {
ip(&format!("link add {INTERFACE} type dummy"));
let _interface = Interface;
ip(&format!("addr add 10.100.0.1/24 dev {INTERFACE}"));
ip(&format!("link set {INTERFACE} up"));
let server = TestServer::start_with(65000, |config| {
config.kernel = Some(FibConfig {
table: TABLE,
metric: 20,
});
})
.await?;
let (base, _http) = server.serve_http().await?;
let mut peer = TestPeer::connect(server.addr).await?;
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 2), 90)
.await?;
// The peer speaks from the loopback and names a router on the test's
// interface as next hop, as a route server would.
let neighbor = Ipv4Addr::new(10, 100, 0, 2);
announce(&mut peer, &["203.0.113.0/24", "198.18.0.0/15"], neighbor).await?;
// One by way of a router that is not a neighbor.
announce(&mut peer, &["192.88.99.0/24"], Ipv4Addr::new(172, 31, 0, 1)).await?;
kernel_becomes(&[
"203.0.113.0/24 via 10.100.0.2 dev ferrix100 proto bgp metric 20",
"198.18.0.0/15 via 10.100.0.2 dev ferrix100 proto bgp metric 20",
]);
let kernel: Value = reqwest::get(format!("{base}/api/kernel"))
.await?
.json()
.await?;
assert_eq!(kernel["enabled"], true);
assert_eq!(kernel["table"], TABLE);
assert_eq!(kernel["installed"], 2);
assert_eq!(
kernel["unresolved"],
serde_json::json!({"next hop not directly connected": 1})
);
let shown = kernel["interfaces"]
.as_array()
.unwrap()
.iter()
.find(|interface| interface["name"] == INTERFACE)
.expect("the test's interface is listed");
assert_eq!(shown["up"], true);
// Its own address, beside the link-local one the kernel gave it.
assert!(
shown["addresses"]
.as_array()
.unwrap()
.contains(&serde_json::json!("10.100.0.1/24")),
"{shown}"
);
// One is withdrawn.
peer.send(&Message::Update(UpdateMessage {
withdrawn_routes: vec![IpPrefix::try_from("198.18.0.0/15").unwrap().into()],
path_attributes: vec![],
announced_routes: vec![],
}))
.await?;
kernel_becomes(&["203.0.113.0/24 via 10.100.0.2 dev ferrix100 proto bgp metric 20"]);
// The session ends: packets must not go on following its routes.
drop(peer);
kernel_becomes(&[]);
Ok(())
}
+46
View File
@@ -171,6 +171,28 @@ impl AttrSet {
&self.wire
}
/// The route's next hop: the one that came in MP_REACH_NLRI, or else
/// the NEXT_HOP attribute. Read off the stored bytes without building
/// anything else, for those who ask it of every route.
pub fn next_hop(&self) -> Option<NlriNextHop> {
let mut plain = None;
let mut wire = &*self.wire;
while let Ok((attribute, rest)) = split_attribute(wire) {
match attribute.type_code {
attr_type::MP_REACH_NLRI => {
// Kept as the next hop's length and the next hop.
let (length, address) = attribute.value.split_first()?;
return NlriNextHop::from_bytes(address)
.filter(|_| usize::from(*length) == address.len());
}
attr_type::NEXT_HOP => plain = NlriNextHop::from_bytes(attribute.value),
_ => {}
}
wire = rest;
}
plain
}
/// Puts the AS numbers of the route's path in `out`, in place of what
/// was there: in the order of the path, the neighbor's first and the
/// origin's last, sets and all. Returns whether the path has a set in
@@ -511,6 +533,30 @@ mod tests {
assert_eq!(store.intern(&attrs, None).unwrap().key().med, Some(7));
}
#[test]
fn the_next_hop_is_read_off_a_set() {
let store = AttrStore::new();
let plain = store.intern(&attributes(vec![3333]), None).unwrap();
assert_eq!(
plain.next_hop(),
Some(NlriNextHop::Ipv4(Ipv4Addr::new(192, 0, 2, 1)))
);
// The one that came with the routes counts, where there is one.
let with_link_local = NlriNextHop::Ipv6WithLl {
global: "2001:db8::1".parse().unwrap(),
link_local: "fe80::1".parse().unwrap(),
};
let v6 = store
.intern(&attributes(vec![3333]), Some(&with_link_local))
.unwrap();
assert_eq!(v6.next_hop(), Some(with_link_local));
// A route of our own making may have none.
let none = store
.intern(&[PathAttribute::Origin(OriginPathAttribute::IGP)], None)
.unwrap();
assert_eq!(none.next_hop(), None);
}
#[test]
fn the_path_is_read_off_a_set_as_it_really_is() {
use bgp_packet::message::{AsPathAttribute, AsPathSegment};
+7
View File
@@ -352,6 +352,13 @@ impl RibTable {
/// with it, and may then be given to another. Whoever was given it must
/// not use it after that: the table cannot tell a late announcement
/// from one of the number's next owner.
/// What is known of the peer with this number, if it is in use. Not to
/// be called from a visitor that is given routes with a shard held.
pub fn peer(&self, index: PeerIndex) -> Option<PeerInfo> {
let peers = self.peers.lock().unwrap_or_else(PoisonError::into_inner);
peers.get(index.as_usize()).copied().flatten()
}
pub fn add_peer(&self, info: PeerInfo) -> Result<PeerIndex, TooManyPeers> {
let mut peers = self.peers.lock().unwrap_or_else(PoisonError::into_inner);
let index = match peers.iter().position(Option::is_none) {
+1
View File
@@ -11,6 +11,7 @@ version.workspace = true
bgp-packet.workspace = true
bgp-primitives.workspace = true
bgp-rib.workspace = true
bgp-fib.workspace = true
bgp-rpki.workspace = true
bytes.workspace = true
clap.workspace = true
+24
View File
@@ -48,6 +48,30 @@ message BgpConfig {
// Route origin validation (RFC 6811) and AS path verification with the
// RPKI. Left out, routes are not validated.
Rpki rpki = 10;
// Put the best route of every prefix in the kernel's routing table, so
// that the machine forwards packets by them. Left out, the daemon only
// speaks BGP and the machine's forwarding is not touched.
Kernel kernel = 11;
}
// Installing routes in the kernel. Linux only; the daemon needs the
// CAP_NET_ADMIN capability.
//
// The best route of a prefix is installed if its next hop is a neighbor:
// an address in the network of one of the machine's interfaces. Routes
// are marked as BGP's ("proto bgp"), and every route with that mark, the
// table and the metric below is taken to be this daemon's: what it finds
// at startup and no longer wants, it removes. When the daemon stops in
// good order it removes its routes.
message Kernel {
// The routing table. Defaults to 254, the main table, which is the one
// packets are forwarded by unless rules say otherwise.
uint32 table = 1;
// The metric of the routes. Defaults to 20. Of several routes for a
// prefix the kernel uses the one with the lowest metric.
uint32 metric = 2;
}
// Where to get the validated ROA payloads that say which AS may originate
+32
View File
@@ -2,6 +2,7 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::str::FromStr;
use std::time::Duration;
use bgp_fib::FibConfig;
use bgp_packet::community;
use bgp_packet::message::OriginPathAttribute;
use bgp_packet::open::AddPathSendRecv;
@@ -43,6 +44,8 @@ pub struct Config {
pub originate: Vec<LocalRoute>,
/// Address for the HTTP API; `None` disables it.
pub http_endpoint: Option<SocketAddr>,
/// Where to install the best routes in the kernel, if at all.
pub kernel: Option<FibConfig>,
/// Where route origin validation gets its data; `None` for no
/// validation.
pub rpki: Option<RpkiConfig>,
@@ -174,6 +177,14 @@ impl Config {
accept_unconfigured_peers: proto.accept_unconfigured_peers,
originate,
http_endpoint,
kernel: proto.kernel.as_ref().map(|kernel| {
let defaults = FibConfig::default();
let or = |value: u32, default: u32| if value == 0 { default } else { value };
FibConfig {
table: or(kernel.table, defaults.table),
metric: or(kernel.metric, defaults.metric),
}
}),
rpki,
})
}
@@ -725,6 +736,27 @@ mod tests {
);
}
#[test]
fn the_kernel_is_left_alone_unless_asked_for() {
assert_eq!(parse(BASE).unwrap().kernel, None);
assert_eq!(
parse(&format!("{BASE}kernel {{}}")).unwrap().kernel,
Some(FibConfig {
table: 254,
metric: 20
})
);
assert_eq!(
parse(&format!("{BASE}kernel {{ table: 1000 metric: 150 }}"))
.unwrap()
.kernel,
Some(FibConfig {
table: 1000,
metric: 150
})
);
}
#[test]
fn route_origin_validation_is_off_unless_caches_are_given() {
assert!(parse(BASE).unwrap().rpki.is_none());
+3
View File
@@ -152,6 +152,9 @@ pub enum ServerError {
/// they panic.
#[error("the {task} task stopped unexpectedly: {reason}")]
TaskStopped { task: &'static str, reason: String },
/// Routes were to be installed in the kernel, and cannot be.
#[error(transparent)]
Kernel(#[from] bgp_fib::FibError),
}
/// Why a configuration was rejected.
+62 -1
View File
@@ -13,6 +13,7 @@ use tokio::net::{TcpListener, TcpSocket, TcpStream};
use tokio::task::{JoinError, JoinHandle};
use tracing::{error, info, warn};
use bgp_fib::{Fib, FibError};
use bgp_rib::table::RibTable;
use bgp_rib::{LocalRib, RibStopped};
use bgp_rpki::Rpki;
@@ -34,6 +35,10 @@ use crate::sync::IgnorePoison;
/// How long sessions get to send their Cease when the server stops.
const SHUTDOWN_GRACE: Duration = Duration::from_secs(2);
/// How long taking our routes out of the kernel may take at shutdown. A
/// full table is some seconds.
const KERNEL_GRACE: Duration = Duration::from_secs(30);
/// How long to wait before accepting again after an accept failed. What
/// makes it fail (no file descriptors left, usually) does not pass at once,
/// and trying again in a loop would only fill the log.
@@ -77,6 +82,9 @@ pub struct BgpServer {
pub rpki: Option<Rpki>,
/// The task that keeps that data up to date.
rpki_client: Option<JoinHandle<()>>,
/// What keeps the best routes in the kernel, where that is configured.
pub kernel: Option<Fib>,
kernel_task: Option<JoinHandle<Result<(), FibError>>>,
}
impl BgpServer {
@@ -110,11 +118,21 @@ impl BgpServer {
// The client starts now: with luck the data is there before the
// first routes are.
let (rpki, rpki_client) = config.rpki.clone().map(Rpki::spawn).unzip();
let rib = Arc::new(RibTable::new(table_shards()));
// Started now, so that a machine on which it cannot be done stops
// the daemon from starting. There is nothing to install yet.
let (kernel, kernel_task) = match config.kernel {
Some(kernel) => Some(Fib::spawn(Arc::clone(&rib), kernel)?),
None => None,
}
.unzip();
Ok(BgpServer {
rpki,
rpki_client,
kernel,
kernel_task,
listeners,
rib: Arc::new(RibTable::new(table_shards())),
rib,
local_rib,
registry: SessionRegistry::with_configured(configured),
refusals: RefusalLog::default(),
@@ -193,6 +211,7 @@ impl BgpServer {
tasks.push(named("dialer", dialer));
}
let mut kernel_task = self.kernel_task;
let connections = ConnectionCounts::default();
let mut refusals = Refusals::default();
let mut accepting: FuturesUnordered<_> = self
@@ -205,6 +224,11 @@ impl BgpServer {
let (listen_addr, result, listener) = tokio::select! {
Some(accepted) = accepting.next() => accepted,
failure = failed(&self.local_rib, &mut tasks) => {
error!(err = %failure, "Shutting down: the server cannot continue");
shut_down(&self.registry, self.kernel.as_ref()).await;
return Err(failure);
}
failure = kernel_failed(&mut kernel_task) => {
error!(err = %failure, "Shutting down: the server cannot continue");
close_sessions(&self.registry).await;
return Err(failure);
@@ -233,6 +257,43 @@ impl BgpServer {
}
}
/// Resolves when the task that keeps the kernel's routes has ended, which
/// while the server runs it does only by failing. Never, if there is no
/// such task.
async fn kernel_failed(task: &mut Option<JoinHandle<Result<(), FibError>>>) -> ServerError {
let Some(running) = task else {
return std::future::pending().await;
};
let ended = running.await;
*task = None;
match ended {
Ok(Err(e)) => e.into(),
Ok(Ok(())) => ServerError::TaskStopped {
task: "kernel routes",
reason: "it was stopped".into(),
},
Err(e) => ServerError::TaskStopped {
task: "kernel routes",
reason: e.to_string(),
},
}
}
/// Leaves in good order: closes every session (see [`close_sessions`]),
/// and then takes our routes out of the kernel, if we put any there. In
/// that order: a peer goes on sending us packets for our routes until it
/// is told we are going.
pub async fn shut_down(registry: &SessionRegistry, kernel: Option<&Fib>) {
close_sessions(registry).await;
if let Some(kernel) = kernel
&& tokio::time::timeout(KERNEL_GRACE, kernel.stop())
.await
.is_err()
{
warn!("Our routes were not all taken out of the kernel in time");
}
}
/// Closes every session politely and waits a moment for them to go: each
/// peer is told we are going (Cease / Administrative Shutdown), so that it
/// stops using our routes now and not when its hold timer expires.
+1
View File
@@ -10,6 +10,7 @@ version.workspace = true
[dependencies]
axum.workspace = true
bgp-fib.workspace = true
bgp-packet.workspace = true
bgp-primitives.workspace = true
bgp-rib.workspace = true
+67
View File
@@ -853,6 +853,73 @@ pub async fn rpki(State(state): State<AppState>) -> Json<RpkiView> {
})
}
/// The best routes in the kernel's routing table: how many are there, and
/// why the others are not.
#[derive(Debug, Clone, Default, Serialize)]
pub struct KernelView {
/// Whether routes are installed in the kernel at all.
pub enabled: bool,
/// The routing table and metric our routes have there.
pub table: Option<u32>,
pub metric: Option<u32>,
/// Whether the kernel has been brought in step with the routing table
/// since the daemon started.
pub synced: bool,
/// How many routes of ours the kernel has.
pub installed: usize,
/// Prefixes whose best route is not in the kernel because its next hop
/// cannot be used, by why.
pub unresolved: BTreeMap<&'static str, usize>,
/// Prefixes whose best route the kernel refused.
pub refused: usize,
/// How many requests the kernel has refused in all, and the last.
pub errors: u64,
pub last_error: Option<String>,
/// The machine's interfaces, as the kernel reports them.
pub interfaces: Vec<InterfaceView>,
}
#[derive(Debug, Clone, Serialize)]
pub struct InterfaceView {
pub name: String,
pub index: u32,
pub up: bool,
/// In CIDR notation, e.g. `192.0.2.1/24`.
pub addresses: Vec<String>,
}
pub async fn kernel(State(state): State<AppState>) -> Json<KernelView> {
let Some(kernel) = &state.kernel else {
return Json(KernelView::default());
};
let (config, status) = (kernel.config(), kernel.status());
Json(KernelView {
enabled: true,
table: Some(config.table),
metric: Some(config.metric),
synced: status.synced,
installed: status.installed,
unresolved: status.unresolved.into_iter().collect(),
refused: status.refused,
errors: status.errors,
last_error: status.last_error,
interfaces: status
.interfaces
.into_iter()
.map(|interface| InterfaceView {
name: interface.name,
index: interface.index,
up: interface.up,
addresses: interface
.addresses
.iter()
.map(|(address, length)| format!("{address}/{length}"))
.collect(),
})
.collect(),
})
}
/// What one peer announced that was refused.
#[derive(Debug, Clone, Serialize)]
pub struct RefusalsView {
+5 -1
View File
@@ -17,6 +17,7 @@ use axum::Router;
use axum::http::header;
use axum::response::{Html, IntoResponse, Redirect};
use axum::routing::get;
use bgp_fib::Fib;
use bgp_rib::LocalRib;
use bgp_rib::table::RibTable;
use bgp_rpki::Rpki;
@@ -26,7 +27,7 @@ use tokio::net::TcpListener;
use tracing::info;
pub use api::{
AttrSummary, ConfigView, LocalRouteView, PeerView, ReceivedRouteView, RefusalsView,
AttrSummary, ConfigView, KernelView, LocalRouteView, PeerView, ReceivedRouteView, RefusalsView,
RefusedRouteView, RoutesPage, RpkiView, StatusView,
};
@@ -90,6 +91,8 @@ pub struct AppState {
pub refusals: RefusalLog,
/// The data for route origin validation, where that is configured.
pub rpki: Option<Rpki>,
/// What keeps the best routes in the kernel, where that is configured.
pub kernel: Option<Fib>,
}
/// The status page, embedded so the binary is self-contained.
@@ -114,6 +117,7 @@ pub fn router(state: AppState) -> Router {
.route("/api/routes/received", get(api::received_routes))
.route("/api/routes/rejected", get(api::rejected_routes))
.route("/api/rpki", get(api::rpki))
.route("/api/kernel", get(api::kernel))
.route("/api/events", get(events::stream))
.route("/metricz", get(metrics::metricz))
.with_state(state)
+23 -3
View File
@@ -49,7 +49,7 @@ Router ID <b id="s-router-id">...</b>, local AS <b id="s-asn">...</b><br>
</div>
<div class="righthand">
Running on <span id="s-hostname">...</span><br>
View: <a href="/metricz">variables</a>, <a href="/api/status">status json</a>, <a href="/api/peers">peers json</a>, <a href="/api/routes/local">originated routes json</a>, <a href="/api/routes/received">received routes json</a>, <a href="/api/routes/rejected">refused routes json</a>, <a href="/api/rpki">rpki json</a><br>
View: <a href="/metricz">variables</a>, <a href="/api/status">status json</a>, <a href="/api/peers">peers json</a>, <a href="/api/routes/local">originated routes json</a>, <a href="/api/routes/received">received routes json</a>, <a href="/api/routes/rejected">refused routes json</a>, <a href="/api/rpki">rpki json</a>, <a href="/api/kernel">kernel json</a><br>
<span id="refresh-ctl"></span><br>
</div>
</div>
@@ -74,6 +74,9 @@ View: <a href="/metricz">variables</a>, <a href="/api/status">status json</a>, <
<tbody id="local-body"></tbody>
</table>
<h1 id="kernel">Kernel routes</h1>
<div id="kernel-body">...</div>
<h1 id="received">Received routes (<span id="received-count">...</span>)</h1>
<form class="lookup" id="lookup">
Look up <input type="text" id="lookup-q" size="30" placeholder="193.0.0.0/21 or 193.0.0.1" title="A prefix, or an address to find every prefix that covers it">
@@ -118,7 +121,7 @@ from <select id="lookup-peer"><option value="">any peer</option></select>
const $ = id => document.getElementById(id);
const esc = s => String(s).replace(/[&<>"]/g, c => ({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;'}[c]));
const state = { status: null, peers: [], local: [], received: [], receivedPage: {}, refused: [], rpki: null };
const state = { status: null, peers: [], local: [], received: [], receivedPage: {}, refused: [], rpki: null, kernel: null };
const DASH = '<span class="dim">-</span>';
// ---- helpers -----------------------------------------------------------
@@ -305,6 +308,22 @@ from <select id="lookup-peer"><option value="">any peer</option></select>
(r.last_error ? `<br><span class="bad">Last error:</span> ${esc(r.last_error)}` : ''));
}
function renderKernel() {
const k = state.kernel, b = $('kernel-body');
if (!k) return;
if (!k.enabled) { setHTML(b, '<span class="dim">Not configured: the best routes are not put in the kernel, and this machine does not forward by them.</span>'); return; }
const unresolved = Object.entries(k.unresolved).map(([why, n]) => `${n} ${esc(why)}`).join(', ');
const interfaces = k.interfaces.map(i =>
`<span class="mono">${esc(i.name)}</span> <span class="${i.up ? 'ok' : 'bad'}">${i.up ? 'up' : 'down'}</span>` +
(i.addresses.length ? ` <span class="mono dim">${i.addresses.map(esc).join(' ')}</span>` : '')).join(' &middot; ');
setHTML(b, `<span class="${k.synced ? 'ok' : 'warn'}">${k.synced ? 'In step' : 'Catching up'}</span> with the routing table: ` +
`<b>${k.installed}</b> routes installed <span class="dim">(table ${k.table}, metric ${k.metric}, proto bgp)</span>` +
(unresolved ? `; <span class="warn">not installed:</span> ${unresolved}` : '') +
(k.refused ? `; <span class="bad">${k.refused} refused by the kernel</span>` : '') + '.' +
(k.last_error ? `<br><span class="bad">Last error</span> (${k.errors} in all): ${esc(k.last_error)}` : '') +
(interfaces ? `<br>Interfaces: ${interfaces}` : ''));
}
// For each peer a line that says how many and why, then its latest few.
const REFUSED_SHOWN = 20;
function renderRefused() {
@@ -356,7 +375,8 @@ from <select id="lookup-peer"><option value="">any peer</option></select>
}
async function loadRefused() { state.refused = await getJSON('/api/routes/rejected'); renderRefused(); }
async function loadRpki() { state.rpki = await getJSON('/api/rpki'); renderRpki(); }
async function loadAll() { await Promise.all([loadStatus(), loadPeers(), loadLocal(), loadReceived(), loadRefused(), loadRpki()]); }
async function loadKernel() { state.kernel = await getJSON('/api/kernel'); renderKernel(); }
async function loadAll() { await Promise.all([loadStatus(), loadPeers(), loadLocal(), loadReceived(), loadRefused(), loadRpki(), loadKernel()]); }
// ---- events ------------------------------------------------------------
function logEvent(html) {