diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e28c428..cc13271 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -26,6 +26,10 @@ jobs: run: cargo clippy --workspace --all-targets -- -D warnings - name: Test run: cargo test --workspace + # The tests that change a kernel's routes, in a network namespace of + # their own; and two daemons that route for each other. + - name: Kernel tests + run: scripts/kernel-tests.sh # A short run of each fuzz target: enough to notice a parser change that # breaks on input the seeds are close to. Long runs are done by hand; see @@ -44,7 +48,7 @@ jobs: env: RUSTFLAGS: "" run: | - for target in message message_body prefix_text config rtr; do + for target in message message_body prefix_text config rtr netlink; do mkdir -p "corpus/$target" cargo fuzz run "$target" "corpus/$target" "seeds/$target" -- -max_total_time=60 done diff --git a/Cargo.lock b/Cargo.lock index 9b104b1..a3d0b39 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -185,10 +185,27 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "bgp-fib" +version = "0.0.2" +dependencies = [ + "bgp-packet", + "bgp-primitives", + "bgp-rib", + "bytes", + "netlink-sys", + "prometheus", + "proptest", + "thiserror 2.0.12", + "tokio", + "tracing", +] + [[package]] name = "bgp-integration-tests" version = "0.0.2" dependencies = [ + "bgp-fib", "bgp-packet", "bgp-primitives", "bgp-rib", @@ -266,6 +283,7 @@ dependencies = [ name = "bgp-server" version = "0.0.2" dependencies = [ + "bgp-fib", "bgp-packet", "bgp-primitives", "bgp-rib", @@ -289,6 +307,7 @@ name = "bgp-web" version = "0.0.2" dependencies = [ "axum", + "bgp-fib", "bgp-packet", "bgp-primitives", "bgp-rib", @@ -961,6 +980,19 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "netlink-sys" +version = "0.8.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd6c30ed10fa69cc491d491b85cc971f6bdeb8e7367b7cde2ee6cc878d583fae" +dependencies = [ + "bytes", + "futures-util", + "libc", + "log", + "tokio", +] + [[package]] name = "nom" version = "8.0.0" diff --git a/Cargo.toml b/Cargo.toml index 1caff74..81ed11a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,6 +27,7 @@ unimplemented = "warn" [workspace.dependencies] # --- Our crates --- +bgp-fib = { path = "crates/fib" } bgp-packet = { path = "crates/packet" } bgp-primitives = { path = "crates/primitives" } bgp-rib = { path = "crates/rib" } diff --git a/bin/bgp/README.md b/bin/bgp/README.md index c2a454b..0266d55 100644 --- a/bin/bgp/README.md +++ b/bin/bgp/README.md @@ -24,6 +24,7 @@ When `http_endpoint` is set the binary also serves: | `GET /api/routes/received` | Routes learned from peers; see "Finding a route" | | `GET /api/routes/rejected` | Routes peers announced that were refused, by peer | | `GET /api/rpki` | Validation with the RPKI: the cache, the data and its age | +| `GET /api/kernel` | The routes in the kernel: how many, and why the others are not; the machine's interfaces | | `GET /api/events` | Server-sent events for session and route changes | | `GET /metricz` | Prometheus metrics | @@ -34,9 +35,11 @@ put it behind a reverse proxy. It establishes sessions (both dialling configured peers and accepting connections, with RFC 4271 collision detection), announces its originated -routes to every peer with itself as next hop, and records what peers send in -an Adj-RIB-In. It does not select best paths, re-advertise received routes or -program a forwarding table, and it has no import or export policy. +routes to every peer with itself as next hop, judges what peers send by each +peer's import policy, keeps what passes, chooses a best route for each +prefix, and on Linux can put those best routes in the kernel so that the +machine forwards by them. It does not yet pass routes learned from one peer +on to another, and has no export policy. ## Who may connect @@ -165,6 +168,67 @@ peers { } ``` +## Forwarding: routes in the kernel (Linux) + +With `kernel {}` in the configuration the daemon puts the best route of every +prefix in the kernel's routing table, over netlink, and keeps it there as the +best routes change. Without it the daemon only speaks BGP and the machine's +forwarding is not touched. + +``` +kernel { + table: 254 # the main table; the default + metric: 20 # the default +} +``` + +- **What is installed.** A best route whose next hop is a neighbor: an + address in the network of one of the machine's interfaces that is up. For + an IPv6 route with a link-local next hop, the interface is the one the + peer is a neighbor on. A route whose next hop is further away (a route + from a peer in our own AS that did not set itself as next hop, say) is not + installed, and neither is whatever it replaced: looking up the way to such + a next hop through other routes is not done yet. The status page counts + these, by reason. +- **What is ours.** Routes are marked `proto bgp`. Every route with that + mark in the configured table with the configured metric is taken to be + this daemon's, and no other route is ever changed or removed. So do not + run two daemons that install BGP routes into the same table with the same + metric. +- **Starting and stopping.** At startup, routes of ours left by an earlier + run are kept if they are still right and removed if not. On SIGINT or + SIGTERM the daemon closes its sessions and then removes its routes. After + a crash they stay until the next start. +- **Interfaces.** The daemon learns the machine's interfaces and addresses + from the kernel and is told when they change; a second later it works all + routes out again, since any next hop may resolve differently. +- **Privileges.** Changing routes needs the `CAP_NET_ADMIN` capability, and + listening on port 179 needs `CAP_NET_BIND_SERVICE`; neither needs root. + Without the first the daemon stops as soon as the kernel refuses it. + + ``` + setcap cap_net_admin,cap_net_bind_service+ep /usr/local/bin/bgp + ``` + +`ip route show proto bgp` shows what is there. `/api/kernel` and the status +page show how many routes are installed, how many are not and why, the last +refusal by the kernel, and the interfaces; `/metricz` has `bgp_kernel_routes`, +`bgp_kernel_routes_unresolved`, `bgp_kernel_routes_refused`, +`bgp_kernel_route_changes_total{change}` and `bgp_kernel_errors_total`. + +To forward packets for others the machine also needs +`net.ipv4.ip_forward=1` and `net.ipv6.conf.all.forwarding=1`; the daemon does +not set them. + +### Testing against a real kernel + +The tests that change a kernel's routes are not run by `cargo test`. +`scripts/kernel-tests.sh` runs them on Linux, each in a network namespace +made for it: the route installer against `ip route`, the daemon with a peer, +and two daemons that route for each other with pings across. From macOS, +`scripts/linux-vm.sh` runs it (or any command, such as `cargo test`) in a +Lima virtual machine that it creates on first use. + ## Finding a route The status page has a look-up box over the received routes, and diff --git a/bin/bgp/src/main.rs b/bin/bgp/src/main.rs index 2f327be..256011c 100644 --- a/bin/bgp/src/main.rs +++ b/bin/bgp/src/main.rs @@ -2,7 +2,7 @@ use std::sync::Arc; use bgp_server::config::Config; use bgp_server::proto::config::BgpConfig; -use bgp_server::server::{BgpServer, close_sessions}; +use bgp_server::server::{BgpServer, shut_down}; use bgp_server::tracing::TracingArgs; use bgp_web::{AppState, RouterInfo}; use clap::Parser; @@ -52,6 +52,7 @@ async fn main() -> Result<()> { rib: Arc::clone(&server.rib), refusals: server.refusals.clone(), rpki: server.rpki.clone(), + kernel: server.kernel.clone(), }; tokio::spawn(async move { if let Err(e) = bgp_web::serve_on(state, listener).await { @@ -63,12 +64,14 @@ async fn main() -> Result<()> { // Leave politely: tell every peer we are going (Cease / Administrative // Shutdown) rather than letting them find a dead TCP connection, which // makes them slower to take us back. - let registry = server.registry.clone(); + // And take our routes out of the kernel: with us gone nothing would + // keep them true. + let (registry, kernel) = (server.registry.clone(), server.kernel.clone()); tokio::select! { result = server.run() => Ok(result?), signal = shutdown_signal() => { info!(signal, "Shutting down, closing BGP sessions"); - close_sessions(®istry).await; + shut_down(®istry, kernel.as_ref()).await; Ok(()) } } diff --git a/crates/fib/Cargo.toml b/crates/fib/Cargo.toml new file mode 100644 index 0000000..70dac35 --- /dev/null +++ b/crates/fib/Cargo.toml @@ -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 diff --git a/crates/fib/src/interfaces.rs b/crates/fib/src/interfaces.rs new file mode 100644 index 0000000..f74f2ad --- /dev/null +++ b/crates/fib/src/interfaces.rs @@ -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, + addresses: HashSet
, +} + +/// `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 { + 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 { + 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 { + 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)); + } +} diff --git a/crates/fib/src/kernel.rs b/crates/fib/src/kernel.rs new file mode 100644 index 0000000..36edcf7 --- /dev/null +++ b/crates/fib/src/kernel.rs @@ -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 = LazyLock::new(|| { + register_int_gauge!( + "bgp_kernel_routes", + "Routes of ours in the kernel's routing table" + ) + .expect("metric registration") +}); + +static UNRESOLVED: LazyLock = 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 = LazyLock::new(|| { + register_int_gauge!( + "bgp_kernel_routes_refused", + "Prefixes whose best route the kernel refused to take" + ) + .expect("metric registration") +}); + +static CHANGES: LazyLock = 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 = 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 { + 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, 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 { + 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, + }, + Withdrawn(IpPrefix), + Synced, +} + +struct Task { + rib: Arc, + 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) -> 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 { 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 = 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, + fib: Fib, + sockets: Sockets, + mut stopping: watch::Receiver, +) -> 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 = 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(()); + } + } + } +} diff --git a/crates/fib/src/lib.rs b/crates/fib/src/lib.rs new file mode 100644 index 0000000..5155d4d --- /dev/null +++ b/crates/fib/src/lib.rs @@ -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, + pub interfaces: Vec, +} + +struct Shared { + config: FibConfig, + status: Mutex, + stop: watch::Sender, + /// Whether the task has ended. + ended: watch::Sender, +} + +/// Handle to the task that keeps the kernel's routes. Cheap to clone. +#[derive(Clone)] +pub struct Fib { + shared: Arc, +} + +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, + config: FibConfig, + ) -> Result<(Self, JoinHandle>), 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, + fib: Fib, + stopping: watch::Receiver, +) -> Result>, 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, + _fib: Fib, + _stopping: watch::Receiver, +) -> Result>, FibError> { + Err(FibError::Unsupported) +} diff --git a/crates/fib/src/netlink.rs b/crates/fib/src/netlink.rs new file mode 100644 index 0000000..2cea1da --- /dev/null +++ b/crates/fib/src/netlink.rs @@ -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, + /// The index of the interface packets leave by. + pub interface: Option, + /// 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 { + 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 { + 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 { + 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::() + .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 { + 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
{ + 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 { + 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 { + ( + any::(), + any::<[u8; 16]>(), + 0u8..=128, + any::(), + any::(), + any::(), + proptest::option::of(any::<[u8; 16]>()), + proptest::option::of(any::()), + ) + .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::(), 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::(), 0..100), + ) { + let wire = message(message_type, 1, &body, &[]); + if let Ok((_, _, used)) = decode(&wire) { + prop_assert!(used > 0 && used <= wire.len()); + } + } + } +} diff --git a/crates/fib/src/plan.rs b/crates/fib/src/plan.rs new file mode 100644 index 0000000..51d5c80 --- /dev/null +++ b/crates/fib/src/plan.rs @@ -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, +) -> Result { + 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, + /// 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, + /// Prefixes whose best route is not installed, and why. + unresolved: HashMap, + /// 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, +} + +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 { + 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) { + 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 { + 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) { + 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) -> Option { + 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 { + 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 { + let stale: Vec = 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 { + 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 { + 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::>(), + [(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), + Withdrawn(u8), + /// A reader that starts over: the whole table again. + Resync, + } + + fn any_step() -> impl Strategy { + 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 = 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> = HashMap::new(); + let apply = |kernel: &mut HashMap, op: Option| match op { + Some(Op::Install { prefix, target }) => { kernel.insert(prefix, target); } + Some(Op::Remove { prefix }) => { kernel.remove(&prefix); } + None => {} + }; + let resolved = |gateway: Option| 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 = 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::(), + plan.unresolved() + ); + } + } + } +} diff --git a/crates/fib/tests/kernel.rs b/crates/fib/tests/kernel.rs new file mode 100644 index 0000000..3475a6f --- /dev/null +++ b/crates/fib/tests/kernel.rs @@ -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, +} + +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>) { + 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 { + 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 { + 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::>().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 = 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 = (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() + ); +} diff --git a/crates/integration-tests/Cargo.toml b/crates/integration-tests/Cargo.toml index 687ef08..da6d3cc 100644 --- a/crates/integration-tests/Cargo.toml +++ b/crates/integration-tests/Cargo.toml @@ -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 diff --git a/crates/integration-tests/src/test_peer.rs b/crates/integration-tests/src/test_peer.rs index e1f94d0..35e8ee8 100644 --- a/crates/integration-tests/src/test_peer.rs +++ b/crates/integration-tests/src/test_peer.rs @@ -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?; diff --git a/crates/integration-tests/src/test_server.rs b/crates/integration-tests/src/test_server.rs index 0673cd0..22fc9f0 100644 --- a/crates/integration-tests/src/test_server.rs +++ b/crates/integration-tests/src/test_server.rs @@ -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, + /// What keeps the best routes in the kernel, if the server does. + pub kernel: Option, 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 })) diff --git a/crates/integration-tests/tests/kernel.rs b/crates/integration-tests/tests/kernel.rs new file mode 100644 index 0000000..bd9fe19 --- /dev/null +++ b/crates/integration-tests/tests/kernel.rs @@ -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 { + 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::>().join(" ")) + .filter(|line| !line.is_empty()) + .collect() +} + +#[track_caller] +fn kernel_becomes(expected: &[&str]) { + let expected: BTreeSet = 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(()) +} diff --git a/crates/rib/src/table/attrs.rs b/crates/rib/src/table/attrs.rs index 5a18b31..63f70cf 100644 --- a/crates/rib/src/table/attrs.rs +++ b/crates/rib/src/table/attrs.rs @@ -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 { + 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}; diff --git a/crates/rib/src/table/mod.rs b/crates/rib/src/table/mod.rs index 86c5f4f..fbffe09 100644 --- a/crates/rib/src/table/mod.rs +++ b/crates/rib/src/table/mod.rs @@ -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 { + 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 { let mut peers = self.peers.lock().unwrap_or_else(PoisonError::into_inner); let index = match peers.iter().position(Option::is_none) { diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index c280e7c..45facd9 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -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 diff --git a/crates/server/proto/config.proto b/crates/server/proto/config.proto index 6d11dba..98ef70a 100644 --- a/crates/server/proto/config.proto +++ b/crates/server/proto/config.proto @@ -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 diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index ceffd57..71534dc 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -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, /// Address for the HTTP API; `None` disables it. pub http_endpoint: Option, + /// Where to install the best routes in the kernel, if at all. + pub kernel: Option, /// Where route origin validation gets its data; `None` for no /// validation. pub rpki: Option, @@ -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()); diff --git a/crates/server/src/error.rs b/crates/server/src/error.rs index c6be0a2..9df3fc1 100644 --- a/crates/server/src/error.rs +++ b/crates/server/src/error.rs @@ -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. diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index a65c8a2..fdcc8ef 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -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, /// The task that keeps that data up to date. rpki_client: Option>, + /// What keeps the best routes in the kernel, where that is configured. + pub kernel: Option, + kernel_task: Option>>, } 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>>) -> 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. diff --git a/crates/web/Cargo.toml b/crates/web/Cargo.toml index b2061b5..19acbfd 100644 --- a/crates/web/Cargo.toml +++ b/crates/web/Cargo.toml @@ -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 diff --git a/crates/web/src/api.rs b/crates/web/src/api.rs index e186e15..000301b 100644 --- a/crates/web/src/api.rs +++ b/crates/web/src/api.rs @@ -853,6 +853,73 @@ pub async fn rpki(State(state): State) -> Json { }) } +/// 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, + pub metric: Option, + /// 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, + /// The machine's interfaces, as the kernel reports them. + pub interfaces: Vec, +} + +#[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, +} + +pub async fn kernel(State(state): State) -> Json { + 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 { diff --git a/crates/web/src/lib.rs b/crates/web/src/lib.rs index 26c97e2..0e1c9f2 100644 --- a/crates/web/src/lib.rs +++ b/crates/web/src/lib.rs @@ -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, + /// What keeps the best routes in the kernel, where that is configured. + pub kernel: Option, } /// 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) diff --git a/crates/web/static/statusz.html b/crates/web/static/statusz.html index 9bf6a65..4888d6b 100644 --- a/crates/web/static/statusz.html +++ b/crates/web/static/statusz.html @@ -49,7 +49,7 @@ Router ID ..., local AS ...
Running on ...
-View: variables, status json, peers json, originated routes json, received routes json, refused routes json, rpki json
+View: variables, status json, peers json, originated routes json, received routes json, refused routes json, rpki json, kernel json

@@ -74,6 +74,9 @@ View: variables, status json, < +

Kernel routes

+
...
+

Received routes (...)

Look up @@ -118,7 +121,7 @@ from const $ = id => document.getElementById(id); const esc = s => String(s).replace(/[&<>"]/g, c => ({'&':'&','<':'<','>':'>','"':'"'}[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 = '-'; // ---- helpers ----------------------------------------------------------- @@ -305,6 +308,22 @@ from (r.last_error ? `
Last error: ${esc(r.last_error)}` : '')); } + function renderKernel() { + const k = state.kernel, b = $('kernel-body'); + if (!k) return; + if (!k.enabled) { setHTML(b, 'Not configured: the best routes are not put in the kernel, and this machine does not forward by them.'); return; } + const unresolved = Object.entries(k.unresolved).map(([why, n]) => `${n} ${esc(why)}`).join(', '); + const interfaces = k.interfaces.map(i => + `${esc(i.name)} ${i.up ? 'up' : 'down'}` + + (i.addresses.length ? ` ${i.addresses.map(esc).join(' ')}` : '')).join(' · '); + setHTML(b, `${k.synced ? 'In step' : 'Catching up'} with the routing table: ` + + `${k.installed} routes installed (table ${k.table}, metric ${k.metric}, proto bgp)` + + (unresolved ? `; not installed: ${unresolved}` : '') + + (k.refused ? `; ${k.refused} refused by the kernel` : '') + '.' + + (k.last_error ? `
Last error (${k.errors} in all): ${esc(k.last_error)}` : '') + + (interfaces ? `
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 } 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) { diff --git a/docs/RFCs.md b/docs/RFCs.md index 5a4ec3a..d971220 100644 --- a/docs/RFCs.md +++ b/docs/RFCs.md @@ -31,7 +31,9 @@ The base protocol specification. | §6.8 | Connection collision detection (BGP Identifier comparison, Cease / Connection Collision Resolution) | ✅ | | §8.1.1 | Optional `CollisionDetectEstablishedState` (`collision_detect_established_state`) | ✅ | | §6.2 | Bad BGP Identifier check (peer using our identifier, or 0.0.0.0) | ✅ | -| §9 | Decision process / best-path selection | ❌ | +| §9 | Decision process / best-path selection | ✅ | +| §9.1.2 | Best routes installed for forwarding (Linux kernel, over netlink), where the next hop is directly connected | ✅ | +| §9.1.2.1 | Routes with an unresolvable next hop excluded from selection; next hops resolved through other routes | ❌ (such a route can be chosen, and is then not installed) | | §9 | Adj-RIB-In (per-peer table of received routes) | ✅ | | §9 | Locally originated routes, advertised to all established peers | ✅ | | §9 | Loc-RIB / Adj-RIB-Out for routes learned from peers (re-advertisement) | ❌ | diff --git a/fuzz/Cargo.toml b/fuzz/Cargo.toml index e4ff614..3969755 100644 --- a/fuzz/Cargo.toml +++ b/fuzz/Cargo.toml @@ -17,6 +17,7 @@ cargo-fuzz = true [dependencies] bgp-packet = { path = "../crates/packet" } bgp-primitives = { path = "../crates/primitives" } +bgp-fib = { path = "../crates/fib" } bgp-rpki = { path = "../crates/rpki" } bgp-server = { path = "../crates/server" } bytes = "1.10.1" @@ -60,3 +61,10 @@ path = "fuzz_targets/rtr.rs" test = false doc = false bench = false + +[[bin]] +name = "netlink" +path = "fuzz_targets/netlink.rs" +test = false +doc = false +bench = false diff --git a/fuzz/README.md b/fuzz/README.md index 7e0c481..5216941 100644 --- a/fuzz/README.md +++ b/fuzz/README.md @@ -13,6 +13,7 @@ you leave them running. | `prefix_text` | A prefix as text | | `config` | The configuration file | | `rtr` | A stream of RTR PDUs, as an RPKI cache sends them | +| `netlink` | Messages from the kernel's routing socket | For the two message targets a finding is either a panic, or a message the decoder accepted that cannot be written and read back as itself. diff --git a/fuzz/fuzz_targets/netlink.rs b/fuzz/fuzz_targets/netlink.rs new file mode 100644 index 0000000..9c6c995 --- /dev/null +++ b/fuzz/fuzz_targets/netlink.rs @@ -0,0 +1,34 @@ +//! Bytes from the kernel's routing socket. The kernel is trusted, but what +//! it sends is read with code of ours, and a daemon that dies on an +//! unusual route stops routing. +#![no_main] + +use bgp_fib::netlink::{Change, Message, decode, route_request}; +use bytes::BytesMut; +use libfuzzer_sys::fuzz_target; + +fuzz_target!(|data: &[u8]| { + let mut rest = data; + // As the daemon reads: one message after another until one is wrong. + while !rest.is_empty() { + let Ok((_, message, used)) = decode(rest) else { + return; + }; + assert!(used > 0 && used <= rest.len()); + rest = &rest[used..]; + let _ = format!("{message:?}"); + // A route that was read can be asked for, and what is asked for + // reads back as itself. + if let Message::Route { present, route } = message { + let change = if present { + Change::Replace + } else { + Change::Remove + }; + let mut out = BytesMut::new(); + route_request(&mut out, 1, change, false, &route); + let (_, again, _) = decode(&out).expect("what is written can be read"); + assert_eq!(again, Message::Route { present, route }); + } + } +}); diff --git a/fuzz/seeds/netlink/interfaces b/fuzz/seeds/netlink/interfaces new file mode 100644 index 0000000..69e5beb Binary files /dev/null and b/fuzz/seeds/netlink/interfaces differ diff --git a/fuzz/seeds/netlink/outcomes b/fuzz/seeds/netlink/outcomes new file mode 100644 index 0000000..ae065aa Binary files /dev/null and b/fuzz/seeds/netlink/outcomes differ diff --git a/fuzz/seeds/netlink/routes b/fuzz/seeds/netlink/routes new file mode 100644 index 0000000..98a7ad3 Binary files /dev/null and b/fuzz/seeds/netlink/routes differ diff --git a/scripts/kernel-tests.sh b/scripts/kernel-tests.sh new file mode 100755 index 0000000..6aa170f --- /dev/null +++ b/scripts/kernel-tests.sh @@ -0,0 +1,38 @@ +#!/bin/sh +# Runs the tests that need a real Linux kernel to change the routes of: +# those of the bgp-fib crate that `cargo test` skips, and the end-to-end +# test of two daemons that route for each other. +# +# They run in a network namespace made for the purpose, so the machine's +# own interfaces and routes are out of their reach, and with the one +# capability the daemon needs, CAP_NET_ADMIN, rather than as root. +# +# Run it on Linux, as a user who can sudo. On another system, +# scripts/linux-vm.sh runs it in a virtual machine. +set -eu + +cd "$(dirname "$0")/.." +[ "$(uname -s)" = Linux ] || { echo "this needs Linux; see scripts/linux-vm.sh" >&2; exit 1; } + +# Built as the user, so that nothing in the target directory is root's. +cargo build --locked -p bgp +cargo test --locked -p bgp-fib --test kernel --no-run +cargo test --locked -p bgp-integration-tests --test kernel --no-run + +namespace="ferrix-test-$$" +sudo modprobe dummy 2>/dev/null || true +sudo ip netns add "$namespace" +trap 'sudo ip netns del "$namespace"' EXIT +sudo ip netns exec "$namespace" ip link set lo up + +# In the namespace, as the user, with CAP_NET_ADMIN and nothing else. +in_namespace() { + sudo ip netns exec "$namespace" \ + setpriv --reuid "$(id -u)" --regid "$(id -g)" --init-groups \ + --inh-caps +net_admin --ambient-caps +net_admin \ + env PATH="$PATH" HOME="$HOME" CARGO_TARGET_DIR="${CARGO_TARGET_DIR:-}" "$@" +} + +in_namespace cargo test --locked -p bgp-fib --test kernel -- --include-ignored --nocapture "$@" +in_namespace cargo test --locked -p bgp-integration-tests --test kernel -- --include-ignored --nocapture "$@" +scripts/two-routers.sh diff --git a/scripts/linux-vm.sh b/scripts/linux-vm.sh new file mode 100755 index 0000000..b50eecc --- /dev/null +++ b/scripts/linux-vm.sh @@ -0,0 +1,34 @@ +#!/bin/sh +# Runs a command in a Linux virtual machine with this repository mounted +# at the same path: for the tests that need a Linux kernel, from a machine +# that is not one. Without arguments it runs scripts/kernel-tests.sh. +# +# scripts/linux-vm.sh the kernel tests +# scripts/linux-vm.sh cargo test the whole test suite, on Linux +# scripts/linux-vm.sh bash a shell +# +# The machine is made on first use with Lima (https://lima-vm.io; on macOS +# `brew install lima`): Ubuntu, with a Rust toolchain. It sees the +# repository read-only and builds into a directory of its own. +# `limactl stop ferrix` stops it and `limactl delete ferrix` removes it. +set -eu + +vm=ferrix +repository="$(cd "$(dirname "$0")/.." && pwd)" +command -v limactl >/dev/null || { echo "limactl not found: install Lima first" >&2; exit 1; } + +if ! limactl list --quiet | grep -qx "$vm"; then + limactl start --name="$vm" --cpus=8 --memory=8 --disk=30 --tty=false template://ubuntu-lts + limactl shell "$vm" bash -c ' + set -eu + sudo apt-get update -qq + sudo DEBIAN_FRONTEND=noninteractive apt-get install -y -qq build-essential iproute2 iputils-ping curl + curl --proto "=https" --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --profile minimal -q + ' +elif [ "$(limactl list --format '{{.Status}}' "$vm")" != Running ]; then + limactl start "$vm" +fi + +[ $# -gt 0 ] || set -- scripts/kernel-tests.sh +limactl shell --workdir "$repository" "$vm" \ + bash -lc 'export CARGO_TARGET_DIR="$HOME/target"; exec "$@"' -- "$@" diff --git a/scripts/two-routers.sh b/scripts/two-routers.sh new file mode 100755 index 0000000..8a4bbfd --- /dev/null +++ b/scripts/two-routers.sh @@ -0,0 +1,144 @@ +#!/bin/sh +# Two daemons that route for each other, on a real kernel: the proof that +# what BGP learns ends up as forwarding. +# +# namespace a namespace b +# lan 10.1.0.1/24 lan 10.2.0.1/24 +# 2001:db8:a::1/48 2001:db8:b::1/48 +# a0 10.0.0.1/24 ------ veth ------ b0 10.0.0.2/24 +# 2001:db8::1/64 2001:db8::2/64 +# +# Each originates its own "lan" prefixes and installs what it learns in +# its kernel. The test passes when each kernel has the other's prefixes as +# BGP routes and a ping from one lan address to the other gets through, +# over IPv4 and IPv6; and when the routes go again as a daemon is stopped. +# +# Run it on Linux as a user who can sudo, after `cargo build -p bgp`; +# scripts/kernel-tests.sh does both. +set -eu + +cd "$(dirname "$0")/.." +bgp="${CARGO_TARGET_DIR:-target}/debug/bgp" +[ -x "$bgp" ] || { echo "$bgp not found: cargo build -p bgp" >&2; exit 1; } + +work="$(mktemp -d)" +a="ferrix-a-$$" +b="ferrix-b-$$" +pids="" +cleanup() { + status=$? + for pid in $pids; do sudo kill "$pid" 2>/dev/null || true; done + sudo ip netns del "$a" 2>/dev/null || true + sudo ip netns del "$b" 2>/dev/null || true + if [ "$status" -ne 0 ]; then + echo "--- FAILED; the daemons said:" >&2 + tail -n 30 "$work"/*.log >&2 || true + fi + rm -rf "$work" + exit "$status" +} +trap cleanup EXIT INT TERM + +in_a() { sudo ip netns exec "$a" "$@"; } +in_b() { sudo ip netns exec "$b" "$@"; } + +sudo modprobe dummy 2>/dev/null || true +sudo ip netns add "$a" +sudo ip netns add "$b" +sudo ip link add a0 netns "$a" type veth peer name b0 netns "$b" +for side in a b; do + case $side in a) n=1; run=in_a ;; b) n=2; run=in_b ;; esac + $run ip link set lo up + $run ip addr add "10.0.0.$n/24" dev "${side}0" + $run ip -6 addr add "2001:db8::$n/64" dev "${side}0" nodad + $run ip link set "${side}0" up + $run ip link add lan type dummy + $run ip addr add "10.$n.0.1/24" dev lan + $run ip -6 addr add "2001:db8:$side::1/48" dev lan nodad + $run ip link set lan up +done + +# One session for each IP version: routes are sent with the session's own +# address as next hop, which has to be of the routes' version. +config() { + side=$1; n=$2; other=$3 + cat > "$work/$side.textpb" < "$work/$side.log" 2>&1 & + pids="$pids $!" +} +start "$a" a +pid_a=$! +start "$b" b + +# Waits until `ip route` in a namespace shows, or no longer shows, a route. +expect() { + run=$1; family=$2; want=$3; route=$4 + for _ in $(seq 100); do + if $run ip "$family" route show proto bgp | sed 's/ */ /g; s/ pref medium//; s/ *$//' | grep -qxF "$route"; then + [ "$want" = present ] && return 0 + else + [ "$want" = absent ] && return 0 + fi + sleep 0.2 + done + echo "expected the route \"$route\" to be $want; the kernel's BGP routes are:" >&2 + $run ip "$family" route show proto bgp >&2 + return 1 +} + +expect in_a -4 present "10.2.0.0/24 via 10.0.0.2 dev a0 metric 20" +expect in_b -4 present "10.1.0.0/24 via 10.0.0.1 dev b0 metric 20" +expect in_a -6 present "2001:db8:b::/48 via 2001:db8::2 dev a0 metric 20" +expect in_b -6 present "2001:db8:a::/48 via 2001:db8::1 dev b0 metric 20" +echo "ok: each kernel has the other's prefixes as BGP routes" + +# From one lan address to the other: there and back by the routes BGP put +# there. +in_a ping -q -c 2 -W 2 -I 10.1.0.1 10.2.0.1 > /dev/null +in_b ping -q -c 2 -W 2 -I 10.2.0.1 10.1.0.1 > /dev/null +in_a ping -q -c 2 -W 2 -I 2001:db8:a::1 2001:db8:b::1 > /dev/null +echo "ok: packets get through, over IPv4 and IPv6" + +# One daemon is stopped in good order. It takes its routes out of its +# kernel; the other, told the session is over, does the same with what it +# learned. +sudo ip netns pids "$a" | xargs sudo kill -TERM +expect in_a -4 absent "10.2.0.0/24 via 10.0.0.2 dev a0 metric 20" +expect in_a -6 absent "2001:db8:b::/48 via 2001:db8::2 dev a0 metric 20" +expect in_b -4 absent "10.1.0.0/24 via 10.0.0.1 dev b0 metric 20" +expect in_b -6 absent "2001:db8:a::/48 via 2001:db8::1 dev b0 metric 20" +if in_b ping -q -c 1 -W 1 -I 10.2.0.1 10.1.0.1 > /dev/null 2>&1; then + echo "packets still get through with the routes gone" >&2 + exit 1 +fi +echo "ok: the routes go when a daemon stops" diff --git a/testdata/server.textpb b/testdata/server.textpb index bfa351a..6ab72cd 100644 --- a/testdata/server.textpb +++ b/testdata/server.textpb @@ -66,6 +66,11 @@ originate { # whose AS path is a leak. Point it at a validator's RTR server. # rpki { caches { address: "127.0.0.1" port: 3323 } } +# Put the best route of every prefix in the kernel's routing table, so that +# this machine forwards by them (Linux; needs CAP_NET_ADMIN). Routes are +# marked "proto bgp"; table and metric default to 254 (main) and 20. +# kernel {} + # Read-only status page, JSON API and metrics. It has no authentication: # keep it on loopback or behind a reverse proxy. http_endpoint { address: "127.0.0.1" port: 8080 }