Rayhaan JaufeerallyandClaude Fable 5.1
befc292215
Readability, no change in behaviour, but for one log line: - The routes a peer holds from us were counted twice over, differently, in the exporter: once per learned-route batch and once per originated route. Learned::held_by_peer counts the learned ones in one place, and ExportLimits::exceeded_by returns the limit that was passed instead of a bool, so that ending the session no longer has to dig it out again with a default that could not happen. - The two hold-downs for a limit passed, ours or the peer's, were two near-identical blocks; they are one match. - How a session ended is put into words in one place (SessionEnd::reason), for the dialer's backoff and the passive peer's state alike. - parse_export_limits parses one block; what a missing block means (no limits at the top, the top's limits for a peer) is said where it differs, instead of Option in and Option out. - The startup warnings about export limits are a function of their own, and the peers without limits are named in one line rather than one line each: a router with eighty sessions does not need eighty warnings. The warning about a peer without a role is gone: the configuration now refuses one. - Hex formatting of attribute octets is one function; the text for an unknown attribute is built in three plain steps. - The exporter's test fixture returns a named struct, not a tuple. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1841 lines
69 KiB
Rust
1841 lines
69 KiB
Rust
//! Announces the routes this router originates to every established
|
|
//! session.
|
|
//!
|
|
//! Each session has an exporter of its own: a task that sends the session
|
|
//! the originated routes and then every change to them. An exporter waits
|
|
//! when its session's queue is full, which is what a peer that reads slowly
|
|
//! or not at all comes to, and it is the only thing that waits: the other
|
|
//! sessions have their own exporters and are not held up.
|
|
//!
|
|
//! An exporter remembers what it has announced. When it has waited so long
|
|
//! that it missed changes, it compares that with the routes as they now
|
|
//! are, announces those, and withdraws what is no longer among them. A peer
|
|
//! that was slow ends up with what every other peer has.
|
|
//!
|
|
//! The advertiser itself only starts an exporter when a session comes up
|
|
//! and stops it when the session goes.
|
|
//!
|
|
//! To a peer above or beside us, the export limits are a check on all of
|
|
//! this ([`ExportLimits`]): routes for prefixes outside them are not
|
|
//! announced, whatever else says they are to be, and if more routes were
|
|
//! to be announced than they allow, none of them is and the session ends.
|
|
//!
|
|
//! To a peer that agreed to graceful restart (RFC 4724) an exporter sends
|
|
//! an End-of-RIB marker for each address family once the peer has all our
|
|
//! routes: at once after the originated ones, or, where learned routes are
|
|
//! passed on too, once the routing table has been gone through and every
|
|
//! peer has sent its own routes since the start. A peer that kept our
|
|
//! routes over a restart of ours drops those it has not had again then.
|
|
|
|
use std::collections::{HashMap, HashSet};
|
|
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use bgp_packet::message::{
|
|
AsPathAttribute, AsPathSegment, CommunitiesPathAttribute, LargeCommunitiesPathAttribute,
|
|
LocalPrefPathAttribute, Message, MpReachNlriPathAttribute, MpUnreachNlriPathAttribute,
|
|
MultiExitDiscPathAttribute, NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage,
|
|
};
|
|
use bgp_primitives::constants::{AfiSafi, IpVersion};
|
|
use bgp_primitives::ip_prefix::{IpPrefix, Nlri};
|
|
use bgp_primitives::parser::ParserContext;
|
|
use bgp_rib::table::{AttrSet, DEFAULT_LOCAL_PREF, Event, PeerIndex, Reader, RibTable};
|
|
use bgp_rib::{LocalRib, LocalRibEvent, LocalRoute};
|
|
use bytes::BytesMut;
|
|
use tokio::sync::broadcast::Receiver;
|
|
use tokio::sync::broadcast::error::RecvError;
|
|
use tokio::sync::{Notify, OwnedSemaphorePermit, Semaphore, watch};
|
|
use tokio::task::JoinHandle;
|
|
use tracing::{debug, error, info, warn};
|
|
|
|
use crate::error::SessionError;
|
|
use crate::export::{self, Egress, Outgoing, Reflection, Source};
|
|
use crate::family::FamilySet;
|
|
use crate::marks::Control;
|
|
use crate::metrics::ROUTES_EXPORTED;
|
|
use crate::policy::PeerRole;
|
|
use crate::session::{PeerId, SessionClosed, SessionEvent, SessionHandle, SessionRegistry};
|
|
|
|
/// Starts the advertiser. It hears of every session that comes up after
|
|
/// this returns, so call it before anything that can bring one up.
|
|
pub fn spawn_advertiser(
|
|
local_rib: LocalRib,
|
|
rib: Arc<RibTable>,
|
|
registry: SessionRegistry,
|
|
local_asn: u32,
|
|
control: Option<Control>,
|
|
cluster_id: Ipv4Addr,
|
|
converged: watch::Receiver<bool>,
|
|
) -> JoinHandle<()> {
|
|
let advertiser = Advertiser {
|
|
converged,
|
|
session_events: registry.subscribe(),
|
|
local_rib,
|
|
rib,
|
|
registry,
|
|
local_asn,
|
|
control,
|
|
cluster_id,
|
|
resyncs: Arc::new(Semaphore::new(RESYNCS_AT_ONCE)),
|
|
exporters: HashMap::new(),
|
|
};
|
|
tokio::spawn(advertiser.run())
|
|
}
|
|
|
|
/// An exporter, as the advertiser holds it.
|
|
struct Running {
|
|
session: SessionHandle,
|
|
/// Tells the exporter to send everything again.
|
|
resend: Arc<Notify>,
|
|
task: JoinHandle<()>,
|
|
}
|
|
|
|
impl Drop for Running {
|
|
fn drop(&mut self) {
|
|
self.task.abort();
|
|
}
|
|
}
|
|
|
|
struct Advertiser {
|
|
local_rib: LocalRib,
|
|
/// The routes learned from peers, of which the best are passed on.
|
|
rib: Arc<RibTable>,
|
|
registry: SessionRegistry,
|
|
local_asn: u32,
|
|
control: Option<Control>,
|
|
/// The cluster this router reflects routes for (RFC 4456).
|
|
cluster_id: Ipv4Addr,
|
|
/// Leave for a session to gather the whole table before it sends it;
|
|
/// see `Learned::gathered`.
|
|
resyncs: Arc<Semaphore>,
|
|
session_events: Receiver<SessionEvent>,
|
|
exporters: HashMap<PeerId, Running>,
|
|
/// Whether every peer has sent its routes since the start.
|
|
converged: watch::Receiver<bool>,
|
|
}
|
|
|
|
impl Advertiser {
|
|
/// Runs for as long as the server does. The server watches for this
|
|
/// task ending: without it nothing is announced.
|
|
async fn run(mut self) {
|
|
loop {
|
|
match self.session_events.recv().await {
|
|
Ok(SessionEvent::Up(session)) => self.start(session),
|
|
Ok(SessionEvent::Down(peer)) => {
|
|
self.exporters.remove(&peer);
|
|
}
|
|
Ok(SessionEvent::RefreshRequested(session)) => {
|
|
if let Some(exporter) = self.exporters.get(&session.peer) {
|
|
debug!(peer = %session.peer, "Route refresh requested, re-sending local routes");
|
|
exporter.resend.notify_one();
|
|
}
|
|
}
|
|
Err(RecvError::Lagged(missed)) => {
|
|
warn!(
|
|
missed,
|
|
"Advertiser lagged on session events, checking every session"
|
|
);
|
|
self.reconcile();
|
|
}
|
|
// The registry is gone, and the server with it.
|
|
Err(RecvError::Closed) => return,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Starts an exporter for a session that has come up.
|
|
fn start(&mut self, session: SessionHandle) {
|
|
let resend = Arc::new(Notify::new());
|
|
let exporter = Exporter {
|
|
// Subscribed here, before the exporter first looks at the
|
|
// routes, so that no change between the two is missed.
|
|
changes: self.local_rib.subscribe(),
|
|
local_rib: self.local_rib.clone(),
|
|
// A follower of the best routes, if the peer is to be told
|
|
// of them: it is given the whole table first.
|
|
learned: session
|
|
.export
|
|
.learned_routes
|
|
.then(|| Learned::new(self.rib.reader())),
|
|
rib: Arc::clone(&self.rib),
|
|
resyncs: Arc::clone(&self.resyncs),
|
|
control: self.control,
|
|
cluster_id: self.cluster_id,
|
|
session: session.clone(),
|
|
local_asn: self.local_asn,
|
|
resend: Arc::clone(&resend),
|
|
announced: HashSet::new(),
|
|
end_of_rib: session
|
|
.graceful_restart
|
|
.as_ref()
|
|
.map(|_| session.families.clone()),
|
|
converged: Some(self.converged.clone()).filter(|converged| !*converged.borrow()),
|
|
};
|
|
let running = Running {
|
|
session: session.clone(),
|
|
resend,
|
|
task: tokio::spawn(exporter.run()),
|
|
};
|
|
// A session with the same peer that this one replaces has gone,
|
|
// whether or not we have heard so yet; its exporter goes with it.
|
|
self.exporters.insert(session.peer, running);
|
|
}
|
|
|
|
/// Makes the exporters match the sessions there are, after events were
|
|
/// missed: sessions may have come and gone unheard, and peers may have
|
|
/// asked for their routes again.
|
|
fn reconcile(&mut self) {
|
|
let sessions = self.registry.list();
|
|
self.exporters.retain(|_, exporter| {
|
|
sessions
|
|
.iter()
|
|
.any(|session| session.is_same_session(&exporter.session))
|
|
});
|
|
for session in sessions {
|
|
match self.exporters.get(&session.peer) {
|
|
Some(exporter) => exporter.resend.notify_one(),
|
|
None => self.start(session),
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Keeps one session's peer up to date with the routes we originate and,
|
|
/// if it is to be told of them, the best of those learned from others.
|
|
struct Exporter {
|
|
rib: Arc<RibTable>,
|
|
resyncs: Arc<Semaphore>,
|
|
control: Option<Control>,
|
|
cluster_id: Ipv4Addr,
|
|
learned: Option<Learned>,
|
|
local_rib: LocalRib,
|
|
changes: Receiver<LocalRibEvent>,
|
|
session: SessionHandle,
|
|
local_asn: u32,
|
|
resend: Arc<Notify>,
|
|
/// The prefixes announced to the peer and not withdrawn since: our
|
|
/// Adj-RIB-Out for it (RFC 4271 §3.2), as far as it needs keeping.
|
|
announced: HashSet<IpPrefix>,
|
|
/// The families to send End-of-RIB markers for, until they are sent;
|
|
/// `None` if graceful restart was not agreed.
|
|
end_of_rib: Option<FamilySet>,
|
|
/// Word of every peer having sent its routes since the start, until it
|
|
/// comes.
|
|
converged: Option<watch::Receiver<bool>>,
|
|
}
|
|
|
|
/// How many changes of best route are taken from the table at a time.
|
|
const EXPORT_BATCH: usize = 65_536;
|
|
|
|
/// How long changes are gathered before they go out, once there are any.
|
|
const COALESCE: Duration = Duration::from_millis(50);
|
|
|
|
/// How many sessions may hold a whole table in hand at once while it is
|
|
/// given out to them anew: some tens of megabytes each for the Internet.
|
|
const RESYNCS_AT_ONCE: usize = 4;
|
|
|
|
/// How many routes ready to go out are remembered, by the attributes they
|
|
/// came with. A table has a few hundred thousand distinct sets.
|
|
const DRESSED_KEPT: usize = 262_144;
|
|
|
|
fn version_index(version: IpVersion) -> usize {
|
|
match version {
|
|
IpVersion::Ipv4 => 0,
|
|
IpVersion::Ipv6 => 1,
|
|
}
|
|
}
|
|
|
|
/// A change of best route, as taken out of the table to be dealt with
|
|
/// once the table is let go of.
|
|
enum News {
|
|
Reset,
|
|
Synced,
|
|
Best {
|
|
prefix: IpPrefix,
|
|
peer: PeerIndex,
|
|
attrs: Arc<AttrSet>,
|
|
},
|
|
Withdrawn(IpPrefix),
|
|
}
|
|
|
|
/// What a learned route goes out as depends on: the attributes it came
|
|
/// with (by where they are), who it came from, and its IP version.
|
|
type DressedKey = (usize, Source, IpVersion);
|
|
|
|
/// A learned route as this session announces it.
|
|
struct Dressed {
|
|
outgoing: Outgoing,
|
|
/// How many octets its attributes take on this session's wire.
|
|
octets: usize,
|
|
}
|
|
|
|
/// What an exporter knows of the learned routes it has passed on.
|
|
struct Learned {
|
|
reader: Reader,
|
|
/// The verdicts so far on a table that is being given out anew, kept
|
|
/// until the table is all there; and the leave to keep them.
|
|
gathered: Option<HashMap<IpPrefix, Option<Arc<Dressed>>>>,
|
|
permit: Option<OwnedSemaphorePermit>,
|
|
/// The prefixes announced to the peer and not withdrawn since.
|
|
announced: HashSet<IpPrefix>,
|
|
/// Whether the peer has been told of the whole table once.
|
|
synced: bool,
|
|
/// While the table is being gone through anew: what was announced
|
|
/// before and has not been heard of again yet.
|
|
stale: Option<HashSet<IpPrefix>>,
|
|
/// Routes as they go out, by the attributes they came with, who they
|
|
/// came from and their IP version: a table's worth of routes share a
|
|
/// few sets of attributes, and each is worked out once. The key holds
|
|
/// the attributes, so that it stays theirs.
|
|
dressed: HashMap<DressedKey, (Arc<AttrSet>, Option<Arc<Dressed>>)>,
|
|
}
|
|
|
|
impl Learned {
|
|
/// How many learned routes the peer holds from us: those announced,
|
|
/// and those not heard of again yet since the table was reset.
|
|
fn held_by_peer(&self) -> usize {
|
|
self.announced.len() + self.stale.as_ref().map_or(0, HashSet::len)
|
|
}
|
|
|
|
fn new(reader: Reader) -> Self {
|
|
Self {
|
|
reader,
|
|
gathered: None,
|
|
permit: None,
|
|
announced: HashSet::new(),
|
|
synced: false,
|
|
stale: None,
|
|
dressed: HashMap::new(),
|
|
}
|
|
}
|
|
|
|
/// The route with `attrs` from `source` as it goes out on `egress`
|
|
/// for prefixes of IP version `version`; `None` if it does not.
|
|
fn dressed(
|
|
&mut self,
|
|
attrs: &Arc<AttrSet>,
|
|
source: Source,
|
|
version: IpVersion,
|
|
egress: &Egress<'_>,
|
|
) -> Option<Arc<Dressed>> {
|
|
let key = (Arc::as_ptr(attrs) as usize, source, version);
|
|
if let Some((_, dressed)) = self.dressed.get(&key) {
|
|
return dressed.clone();
|
|
}
|
|
let dressed = attrs
|
|
.attributes()
|
|
.and_then(|(attributes, next_hop)| {
|
|
egress.outgoing(attributes, next_hop, source, version)
|
|
})
|
|
.map(|outgoing| {
|
|
// As long as they will be on the wire, to know how many
|
|
// prefixes fit beside them.
|
|
let ctx = ParserContext {
|
|
four_octet_asn: egress.towards.four_octet_asn,
|
|
ibgp: egress.towards.ibgp,
|
|
add_path: Vec::new(),
|
|
};
|
|
let mut wire = BytesMut::new();
|
|
let mut attributes = outgoing.attributes.clone();
|
|
export::finish(&mut attributes, &egress.towards);
|
|
let octets = attributes
|
|
.iter()
|
|
.try_for_each(|attribute| attribute.to_wire(&ctx, &mut wire))
|
|
.map_or(usize::MAX, |()| wire.len());
|
|
Arc::new(Dressed { outgoing, octets })
|
|
})
|
|
// One that cannot be written is one that does not go.
|
|
.filter(|dressed| dressed.octets < 3072);
|
|
if self.dressed.len() >= DRESSED_KEPT {
|
|
self.dressed.clear();
|
|
}
|
|
self.dressed
|
|
.insert(key, (Arc::clone(attrs), dressed.clone()));
|
|
dressed
|
|
}
|
|
}
|
|
|
|
/// Waits until every peer has sent its routes since the start; for ever if
|
|
/// that is not to be waited for.
|
|
async fn converged(converged: &mut Option<watch::Receiver<bool>>) {
|
|
match converged {
|
|
// If the sender is gone, so is the server: nothing to wait for.
|
|
Some(converged) => {
|
|
let _ = converged.wait_for(|done| *done).await;
|
|
}
|
|
None => std::future::pending().await,
|
|
}
|
|
}
|
|
|
|
/// Waits until there may be changes of best route to pass on; for ever if
|
|
/// the peer is not told of them.
|
|
async fn changed(learned: &mut Option<Learned>) {
|
|
match learned {
|
|
Some(learned) => learned.reader.changed().await,
|
|
None => std::future::pending().await,
|
|
}
|
|
}
|
|
|
|
/// Why an exporter has nothing more to do.
|
|
struct Finished;
|
|
|
|
impl From<SessionClosed> for Finished {
|
|
fn from(_: SessionClosed) -> Self {
|
|
Self
|
|
}
|
|
}
|
|
|
|
impl Exporter {
|
|
/// Runs until the session closes or the local RIB stops; the
|
|
/// advertiser stops it when it hears the session has gone.
|
|
async fn run(mut self) {
|
|
info!(peer = %self.session.peer, "Session up, sending local routes");
|
|
let _ = self.export().await;
|
|
}
|
|
|
|
async fn export(&mut self) -> Result<(), Finished> {
|
|
self.send_all().await?;
|
|
self.end_of_rib().await?;
|
|
loop {
|
|
tokio::select! {
|
|
change = self.changes.recv() => match change {
|
|
Ok(LocalRibEvent::Announced(route)) => self.announce(&route).await?,
|
|
Ok(LocalRibEvent::Withdrawn { prefix }) => self.withdraw(prefix).await?,
|
|
Err(RecvError::Lagged(missed)) => {
|
|
info!(
|
|
peer = %self.session.peer,
|
|
missed,
|
|
"Peer fell behind on local routes, bringing it up to date"
|
|
);
|
|
self.send_all().await?;
|
|
}
|
|
Err(RecvError::Closed) => return Err(Finished),
|
|
},
|
|
() = self.resend.notified() => {
|
|
self.send_all().await?;
|
|
// Everything, again: a new follower starts with the
|
|
// whole table.
|
|
if let Some(learned) = &mut self.learned {
|
|
learned.reader = self.rib.reader();
|
|
}
|
|
}
|
|
() = changed(&mut self.learned) => {
|
|
// Changes come in runs: a table arriving, a peer
|
|
// going. Gathered for a moment, the routes that go out
|
|
// alike go in one message, and the whole run costs one
|
|
// look at what each peer is told.
|
|
tokio::time::sleep(COALESCE).await;
|
|
self.follow().await?;
|
|
self.end_of_rib().await?;
|
|
}
|
|
() = converged(&mut self.converged), if self.converged.is_some() => {
|
|
self.converged = None;
|
|
self.end_of_rib().await?;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Sends the End-of-RIB markers, if they are still to be sent and the
|
|
/// peer has had all our routes: the originated ones, and the learned
|
|
/// ones as far as there are to be any.
|
|
async fn end_of_rib(&mut self) -> Result<(), Finished> {
|
|
let complete = match &self.learned {
|
|
None => true,
|
|
Some(learned) => learned.synced && self.converged.is_none(),
|
|
};
|
|
if !complete {
|
|
return Ok(());
|
|
}
|
|
let Some(families) = self.end_of_rib.take() else {
|
|
return Ok(());
|
|
};
|
|
debug!(peer = %self.session.peer, %families, "Sending End-of-RIB");
|
|
for family in families.iter() {
|
|
self.session
|
|
.send(Message::Update(UpdateMessage::end_of_rib(family)))
|
|
.await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// The session as far as it decides what goes out on it and how.
|
|
fn towards(&self) -> Towards {
|
|
let session = &self.session;
|
|
Towards {
|
|
local_asn: self.local_asn,
|
|
next_hop: session.local_addr.ip(),
|
|
four_octet_asn: session.four_octet_asn,
|
|
ibgp: session.ibgp,
|
|
role: session.role,
|
|
extended_next_hop: session.extended_next_hop,
|
|
}
|
|
}
|
|
|
|
/// Takes the next changes of best route from the table and tells the
|
|
/// peer of those that are for it.
|
|
async fn follow(&mut self) -> Result<(), Finished> {
|
|
let towards = self.towards();
|
|
let Some(learned) = &mut self.learned else {
|
|
return Ok(());
|
|
};
|
|
// Only what is needed is taken while the table is held; and all
|
|
// of what there is, up to a limit, so that it goes out together.
|
|
let mut news = Vec::new();
|
|
while news.len() < EXPORT_BATCH {
|
|
let progress = learned.reader.poll(EXPORT_BATCH - news.len(), |event| {
|
|
news.push(match event {
|
|
Event::Reset => News::Reset,
|
|
Event::Synced => News::Synced,
|
|
Event::Best {
|
|
prefix,
|
|
peer,
|
|
attrs,
|
|
..
|
|
} => News::Best {
|
|
prefix,
|
|
peer,
|
|
attrs: Arc::clone(attrs),
|
|
},
|
|
Event::Withdrawn { prefix } => News::Withdrawn(prefix),
|
|
// Not given to a follower of the best routes.
|
|
Event::Routes { .. } => return,
|
|
});
|
|
});
|
|
if progress.caught_up || progress.events == 0 {
|
|
break;
|
|
}
|
|
}
|
|
|
|
let session = &self.session;
|
|
let limits = session.export_limits.as_deref();
|
|
let egress = Egress {
|
|
towards,
|
|
peer_asn: session.peer.asn,
|
|
rib_peer: session.rib_peer,
|
|
policy: &session.export,
|
|
groups: &session.groups,
|
|
control: self.control,
|
|
reflection: Reflection {
|
|
cluster_id: self.cluster_id,
|
|
client: session.reflector_client,
|
|
peer_router_id: session.peer_router_id,
|
|
},
|
|
};
|
|
// What the peer is to end up with for each prefix heard of: the
|
|
// last word on it counts. While the whole table is being given out
|
|
// anew, which is shard by shard, this is kept until it is all
|
|
// there: the routes that share attributes are in every shard, and
|
|
// are only sent together if gathered first.
|
|
let mut verdicts = learned.gathered.take().unwrap_or_default();
|
|
let mut synced = false;
|
|
// Who the routes are from, asked of the table once for each.
|
|
let mut sources: HashMap<PeerIndex, Option<Source>> = HashMap::new();
|
|
for item in news {
|
|
match item {
|
|
News::Reset => {
|
|
// Whatever was announced and is not heard of again by
|
|
// the time the table is through is withdrawn then.
|
|
let known = std::mem::take(&mut learned.announced);
|
|
learned.stale.get_or_insert_default().extend(known);
|
|
verdicts.clear();
|
|
synced = false;
|
|
// Gathering a table takes memory: only so many sessions
|
|
// do it at once, the others wait their turn.
|
|
if learned.permit.is_none() {
|
|
learned.permit = self.resyncs.clone().acquire_owned().await.ok();
|
|
}
|
|
}
|
|
News::Synced => {
|
|
synced = true;
|
|
learned.permit = None;
|
|
}
|
|
News::Withdrawn(prefix) => {
|
|
verdicts.insert(prefix, None);
|
|
}
|
|
News::Best {
|
|
prefix,
|
|
peer,
|
|
attrs,
|
|
} => {
|
|
let source = *sources.entry(peer).or_insert_with(|| {
|
|
self.rib.peer(peer).map(|info| Source {
|
|
peer,
|
|
ibgp: info.ibgp,
|
|
reflector_client: info.reflector_client,
|
|
router_id: info.router_id,
|
|
})
|
|
});
|
|
let dressed = match source {
|
|
// What we originate ourselves is announced as
|
|
// that, and not as someone else's route for it.
|
|
Some(source)
|
|
if session.families.carries(&prefix)
|
|
&& !self.announced.contains(&prefix)
|
|
&& limits.is_none_or(|limits| limits.allows(&prefix)) =>
|
|
{
|
|
learned.dressed(&attrs, source, prefix.address_family(), &egress)
|
|
}
|
|
// The peer has gone, and its routes with it.
|
|
_ => None,
|
|
};
|
|
verdicts.insert(prefix, dressed);
|
|
}
|
|
}
|
|
}
|
|
|
|
if learned.permit.is_some() && !synced {
|
|
// Not all there yet.
|
|
learned.gathered = Some(verdicts);
|
|
return Ok(());
|
|
}
|
|
|
|
// Withdrawals first, then announcements, routes that go out alike
|
|
// together.
|
|
let mut withdrawn: [Vec<Nlri>; 2] = Default::default();
|
|
let mut announced: HashMap<(usize, IpVersion), (Arc<Dressed>, Vec<Nlri>)> = HashMap::new();
|
|
for (prefix, verdict) in verdicts {
|
|
let version = prefix.address_family();
|
|
// Heard of again: no longer in doubt, whichever way it goes.
|
|
let was_stale = learned
|
|
.stale
|
|
.as_mut()
|
|
.is_some_and(|stale| stale.remove(&prefix));
|
|
match verdict {
|
|
Some(dressed) => {
|
|
learned.announced.insert(prefix);
|
|
announced
|
|
.entry((Arc::as_ptr(&dressed) as usize, version))
|
|
.or_insert_with(|| (dressed, Vec::new()))
|
|
.1
|
|
.push(prefix.into());
|
|
}
|
|
None => {
|
|
if learned.announced.remove(&prefix) || was_stale {
|
|
withdrawn[version_index(version)].push(prefix.into());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if synced {
|
|
learned.synced = true;
|
|
}
|
|
if synced && let Some(stale) = learned.stale.take() {
|
|
for prefix in stale {
|
|
withdrawn[version_index(prefix.address_family())].push(prefix.into());
|
|
}
|
|
}
|
|
// What the peer would hold once this goes out: none of it goes if
|
|
// that is more than it may.
|
|
let routes = self.announced.len() + learned.held_by_peer();
|
|
if let Some(limit) = limits.and_then(|limits| limits.exceeded_by(routes)) {
|
|
return Err(over_limit(session, limit, routes));
|
|
}
|
|
for (version, routes) in [IpVersion::Ipv4, IpVersion::Ipv6]
|
|
.into_iter()
|
|
.zip(withdrawn)
|
|
{
|
|
if routes.is_empty() {
|
|
continue;
|
|
}
|
|
ROUTES_EXPORTED
|
|
.with_label_values(&["withdrawn"])
|
|
.inc_by(routes.len() as u64);
|
|
for update in export::withdrawals(version, &routes, &towards) {
|
|
session.send(Message::Update(update)).await?;
|
|
}
|
|
}
|
|
for ((_, version), (dressed, routes)) in announced {
|
|
ROUTES_EXPORTED
|
|
.with_label_values(&["announced"])
|
|
.inc_by(routes.len() as u64);
|
|
let updates = export::announcements(
|
|
&dressed.outgoing,
|
|
version,
|
|
&routes,
|
|
&towards,
|
|
dressed.octets,
|
|
);
|
|
for update in updates {
|
|
session.send(Message::Update(update)).await?;
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Brings the peer to the routes as they now are: announces every one
|
|
/// of them, and withdraws what was announced and is no longer there.
|
|
///
|
|
/// Changes made while this runs are in `changes`, to be handled after.
|
|
/// Some of those may be older than the routes read here; handling them
|
|
/// again, in order, ends in the same place.
|
|
async fn send_all(&mut self) -> Result<(), Finished> {
|
|
let routes = self.local_rib.snapshot().await.map_err(|_| Finished)?;
|
|
let current: HashSet<IpPrefix> = routes.iter().map(|route| route.prefix).collect();
|
|
let gone: Vec<IpPrefix> = self.announced.difference(¤t).copied().collect();
|
|
for prefix in gone {
|
|
self.withdraw(prefix).await?;
|
|
}
|
|
for route in &routes {
|
|
self.announce(route).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
async fn announce(&mut self, route: &LocalRoute) -> Result<(), Finished> {
|
|
let session = &self.session;
|
|
if !session.families.carries(&route.prefix) {
|
|
debug!(
|
|
peer = %session.peer,
|
|
prefix = %route.prefix,
|
|
"Skipping route: its address family was not negotiated"
|
|
);
|
|
return Ok(());
|
|
}
|
|
let limits = session.export_limits.as_deref();
|
|
if limits.is_some_and(|limits| !limits.allows(&route.prefix)) {
|
|
debug!(
|
|
peer = %session.peer,
|
|
prefix = %route.prefix,
|
|
"Not announcing route: its prefix is not in the peer's export limits"
|
|
);
|
|
return Ok(());
|
|
}
|
|
// What the peer would hold with this route.
|
|
let routes = self.announced.len()
|
|
+ usize::from(!self.announced.contains(&route.prefix))
|
|
+ self.learned.as_ref().map_or(0, Learned::held_by_peer);
|
|
if let Some(limit) = limits.and_then(|limits| limits.exceeded_by(routes)) {
|
|
return Err(over_limit(session, limit, routes));
|
|
}
|
|
let towards = self.towards();
|
|
let Some(mut update) = build_announce(route, &towards) else {
|
|
debug!(
|
|
peer = %session.peer,
|
|
prefix = %route.prefix,
|
|
"Skipping route: address family does not match session transport"
|
|
);
|
|
return Ok(());
|
|
};
|
|
// What every route on its way to this peer goes through: whether
|
|
// its communities let it go there, and what it is marked with.
|
|
let egress = Egress {
|
|
towards,
|
|
peer_asn: session.peer.asn,
|
|
rib_peer: session.rib_peer,
|
|
policy: &session.export,
|
|
groups: &session.groups,
|
|
control: self.control,
|
|
reflection: Reflection {
|
|
cluster_id: self.cluster_id,
|
|
client: session.reflector_client,
|
|
peer_router_id: session.peer_router_id,
|
|
},
|
|
};
|
|
if !egress.dress(&mut update.path_attributes) {
|
|
debug!(
|
|
peer = %session.peer,
|
|
prefix = %route.prefix,
|
|
"Not announcing route: its communities say not to this peer"
|
|
);
|
|
return Ok(());
|
|
}
|
|
export::finish(&mut update.path_attributes, &towards);
|
|
debug!(peer = %session.peer, prefix = %route.prefix, "Announcing local route");
|
|
session.send(Message::Update(update)).await?;
|
|
self.announced.insert(route.prefix);
|
|
Ok(())
|
|
}
|
|
|
|
async fn withdraw(&mut self, prefix: IpPrefix) -> Result<(), Finished> {
|
|
if !self.announced.remove(&prefix) {
|
|
// The peer was never told of it.
|
|
return Ok(());
|
|
}
|
|
debug!(peer = %self.session.peer, %prefix, "Withdrawing local route");
|
|
self.session
|
|
.send(Message::Update(build_withdraw(
|
|
&prefix,
|
|
self.session.extended_next_hop,
|
|
)))
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// Ends `session` because `routes` routes were to be announced on it, over
|
|
/// its export limit of `limit`; none of them has been.
|
|
fn over_limit(session: &SessionHandle, limit: u32, routes: usize) -> Finished {
|
|
error!(
|
|
peer = %session.peer,
|
|
name = session.name.as_deref().unwrap_or_default(),
|
|
routes,
|
|
limit,
|
|
"More routes were to be advertised to the peer than its export limit allows; advertising \
|
|
none of them, ending the session and keeping the peer down"
|
|
);
|
|
session.end(SessionError::ExportLimitExceeded { limit, routes });
|
|
Finished
|
|
}
|
|
|
|
/// What about a session decides how a route is announced on it.
|
|
#[derive(Debug, Clone, Copy)]
|
|
pub struct Towards {
|
|
pub local_asn: u32,
|
|
/// Our address on the session, which is the routes' next hop.
|
|
pub next_hop: IpAddr,
|
|
/// Whether the session negotiated four-octet AS numbers. If it did
|
|
/// not, and an AS number in the path needs four octets, the path is
|
|
/// also sent as AS4_PATH so that it survives the two-octet speaker
|
|
/// (RFC 6793 §4.2.2).
|
|
pub four_octet_asn: bool,
|
|
/// Whether the peer is in our own AS.
|
|
pub ibgp: bool,
|
|
/// What the peer is to us, if that is known.
|
|
pub role: Option<PeerRole>,
|
|
/// Whether the session takes IPv4 routes with IPv6 next hops
|
|
/// (RFC 8950).
|
|
pub extended_next_hop: bool,
|
|
}
|
|
|
|
/// Builds the UPDATE announcing `route` on a session.
|
|
///
|
|
/// Returns `None` when the route cannot be given the session's address as
|
|
/// next hop: an IPv6 route on a session over IPv4, or an IPv4 route on a
|
|
/// session over IPv6 that did not agree on RFC 8950.
|
|
pub fn build_announce(route: &LocalRoute, towards: &Towards) -> Option<UpdateMessage> {
|
|
let attrs = &route.attrs;
|
|
|
|
// Our AS number goes on the path as a route leaves the AS, and not
|
|
// before (RFC 4271 §5.1.2): a route we originate has an empty path
|
|
// inside it. A peer in our AS that found our AS number on the path
|
|
// would take the route for one that had looped.
|
|
let as_path = AsPathAttribute {
|
|
segments: if towards.ibgp {
|
|
vec![]
|
|
} else {
|
|
vec![AsPathSegment {
|
|
ordered: true,
|
|
path: vec![towards.local_asn; 1 + usize::from(attrs.prepend)],
|
|
}]
|
|
},
|
|
};
|
|
let as4_path = if towards.four_octet_asn {
|
|
None
|
|
} else {
|
|
as_path.as4_path()
|
|
};
|
|
|
|
// Path attributes in ascending type-code order, as most implementations
|
|
// expect them.
|
|
let mut path_attributes = vec![
|
|
PathAttribute::Origin(attrs.origin),
|
|
PathAttribute::AsPath(as_path),
|
|
];
|
|
// IPv4 routes go in the body of the UPDATE with their next hop in
|
|
// NEXT_HOP; IPv6 routes go in MP_REACH_NLRI, which carries its own.
|
|
let mut announced_routes = vec![];
|
|
let mp_reach = match (route.prefix.address_family(), towards.next_hop) {
|
|
(IpVersion::Ipv4, IpAddr::V4(next_hop)) => {
|
|
path_attributes.push(PathAttribute::NextHop(NextHopPathAttribute(next_hop)));
|
|
announced_routes.push(route.prefix.into());
|
|
None
|
|
}
|
|
(IpVersion::Ipv6, IpAddr::V6(next_hop)) => {
|
|
Some(PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
|
|
family: AfiSafi::IPV6_UNICAST,
|
|
next_hop: ipv6_next_hop(next_hop),
|
|
prefixes: vec![route.prefix.into()],
|
|
}))
|
|
}
|
|
// An IPv4 route over a link that has only IPv6 addresses: in
|
|
// MP_REACH_NLRI like an IPv6 route, with the IPv6 next hop
|
|
// (RFC 8950 §4).
|
|
(IpVersion::Ipv4, IpAddr::V6(next_hop)) if towards.extended_next_hop => {
|
|
Some(PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
|
|
family: AfiSafi::IPV4_UNICAST,
|
|
next_hop: ipv6_next_hop(next_hop),
|
|
prefixes: vec![route.prefix.into()],
|
|
}))
|
|
}
|
|
_ => return None,
|
|
};
|
|
if let Some(med) = attrs.med {
|
|
path_attributes.push(PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(
|
|
med,
|
|
)));
|
|
}
|
|
// LOCAL_PREF is sent to peers in our AS and to no others (RFC 4271
|
|
// §5.1.5).
|
|
if towards.ibgp {
|
|
path_attributes.push(PathAttribute::LocalPref(LocalPrefPathAttribute(
|
|
DEFAULT_LOCAL_PREF,
|
|
)));
|
|
}
|
|
if !attrs.communities.is_empty() {
|
|
path_attributes.push(PathAttribute::Communities(CommunitiesPathAttribute(
|
|
attrs.communities.clone(),
|
|
)));
|
|
}
|
|
path_attributes.extend(mp_reach);
|
|
path_attributes.extend(as4_path);
|
|
if !attrs.large_communities.is_empty() {
|
|
path_attributes.push(PathAttribute::LargeCommunities(
|
|
LargeCommunitiesPathAttribute {
|
|
communities: attrs.large_communities.clone(),
|
|
},
|
|
));
|
|
}
|
|
// A route that goes down to a customer or across to a peer may not
|
|
// come up again from there: it is marked, for every network after us
|
|
// to see (RFC 9234 §5, egress, 1). Routes we originate carry no such
|
|
// mark of anyone else's. When routes learned from peers are passed
|
|
// on, one that carries the mark must not go to a provider, a peer or
|
|
// a route server at all (egress, 2;
|
|
// `PeerRole::may_be_sent_only_to_customer_routes`).
|
|
if let Some(marked_by) = towards
|
|
.role
|
|
.filter(|_| !towards.ibgp)
|
|
.and_then(|role| role.only_to_customer_on_export(towards.local_asn))
|
|
{
|
|
path_attributes.push(PathAttribute::OnlyToCustomer(marked_by));
|
|
}
|
|
|
|
Some(UpdateMessage {
|
|
withdrawn_routes: vec![],
|
|
path_attributes,
|
|
announced_routes,
|
|
})
|
|
}
|
|
|
|
/// Our address `ours` as the next hop of a route.
|
|
///
|
|
/// A link-local address, which is all there is on a session between such
|
|
/// addresses, goes in both of the places MP_REACH_NLRI has: where the
|
|
/// global address would be and where the link-local one is (RFC 2545 §3
|
|
/// has no form for a link-local address alone). FRR sends it so, and takes
|
|
/// the route for unreachable if it comes by itself; BIRD takes either.
|
|
pub(crate) fn ipv6_next_hop(ours: Ipv6Addr) -> NlriNextHop {
|
|
if ours.is_unicast_link_local() {
|
|
NlriNextHop::Ipv6WithLl {
|
|
global: ours,
|
|
link_local: ours,
|
|
}
|
|
} else {
|
|
NlriNextHop::Ipv6(ours)
|
|
}
|
|
}
|
|
|
|
/// Builds the UPDATE withdrawing `prefix`. With `extended_next_hop` the
|
|
/// session has IPv4 routes announced in MP_REACH_NLRI (RFC 8950), and they
|
|
/// are withdrawn the way they were announced.
|
|
pub fn build_withdraw(prefix: &IpPrefix, extended_next_hop: bool) -> UpdateMessage {
|
|
match prefix.address_family() {
|
|
IpVersion::Ipv4 if extended_next_hop => UpdateMessage {
|
|
withdrawn_routes: vec![],
|
|
path_attributes: vec![PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute {
|
|
family: AfiSafi::IPV4_UNICAST,
|
|
prefixes: vec![Nlri::from(*prefix)],
|
|
})],
|
|
announced_routes: vec![],
|
|
},
|
|
IpVersion::Ipv4 => UpdateMessage {
|
|
withdrawn_routes: vec![Nlri::from(*prefix)],
|
|
path_attributes: vec![],
|
|
announced_routes: vec![],
|
|
},
|
|
IpVersion::Ipv6 => UpdateMessage {
|
|
withdrawn_routes: vec![],
|
|
path_attributes: vec![PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute {
|
|
family: AfiSafi::IPV6_UNICAST,
|
|
prefixes: vec![Nlri::from(*prefix)],
|
|
})],
|
|
announced_routes: vec![],
|
|
},
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use std::net::{Ipv4Addr, Ipv6Addr};
|
|
|
|
use bgp_packet::message::{LargeCommunity, OriginPathAttribute};
|
|
use bgp_primitives::parser::ParserContext;
|
|
use bgp_rib::LocalRouteAttrs;
|
|
use bytes::BytesMut;
|
|
|
|
use super::*;
|
|
|
|
fn prefix(s: &str) -> IpPrefix {
|
|
IpPrefix::try_from(s).unwrap()
|
|
}
|
|
|
|
/// On a session between link-local addresses ours is the next hop,
|
|
/// written the way FRR writes its own and takes ours.
|
|
#[test]
|
|
fn a_link_local_next_hop_is_given_twice() {
|
|
let ours: Ipv6Addr = "fe80::1".parse().unwrap();
|
|
let towards = Towards {
|
|
extended_next_hop: true,
|
|
..towards(65000, IpAddr::V6(ours), true)
|
|
};
|
|
for (route, family) in [
|
|
("10.1.0.0/16", AfiSafi::IPV4_UNICAST),
|
|
("2001:db8:1::/48", AfiSafi::IPV6_UNICAST),
|
|
] {
|
|
let local = LocalRoute::new(prefix(route), LocalRouteAttrs::default());
|
|
let update = build_announce(&local, &towards).unwrap();
|
|
assert!(update.path_attributes.contains(&PathAttribute::MpReachNlri(
|
|
MpReachNlriPathAttribute {
|
|
family,
|
|
next_hop: NlriNextHop::Ipv6WithLl {
|
|
global: ours,
|
|
link_local: ours,
|
|
},
|
|
prefixes: vec![prefix(route).into()],
|
|
}
|
|
)));
|
|
}
|
|
}
|
|
|
|
/// RFC 8950: over a link that has only IPv6 addresses, an IPv4 route
|
|
/// goes in MP_REACH_NLRI with our IPv6 address as next hop, and is
|
|
/// withdrawn in MP_UNREACH_NLRI.
|
|
#[test]
|
|
fn an_ipv4_route_goes_over_an_ipv6_session_if_both_agreed() {
|
|
let route = LocalRoute::new(prefix("10.1.0.0/16"), LocalRouteAttrs::default());
|
|
let ours: Ipv6Addr = "2001:db8::1".parse().unwrap();
|
|
let over_ipv6 = towards(65000, IpAddr::V6(ours), true);
|
|
// Not agreed: the route has no next hop the peer could use.
|
|
assert!(build_announce(&route, &over_ipv6).is_none());
|
|
|
|
let agreed = Towards {
|
|
extended_next_hop: true,
|
|
..over_ipv6
|
|
};
|
|
let update = build_announce(&route, &agreed).unwrap();
|
|
assert!(update.announced_routes.is_empty(), "not in the body");
|
|
assert!(
|
|
!update
|
|
.path_attributes
|
|
.iter()
|
|
.any(|attribute| matches!(attribute, PathAttribute::NextHop(_))),
|
|
"no NEXT_HOP attribute"
|
|
);
|
|
assert!(update.path_attributes.contains(&PathAttribute::MpReachNlri(
|
|
MpReachNlriPathAttribute {
|
|
family: AfiSafi::IPV4_UNICAST,
|
|
next_hop: NlriNextHop::Ipv6(ours),
|
|
prefixes: vec![prefix("10.1.0.0/16").into()],
|
|
}
|
|
)));
|
|
// What is written is what a peer reads.
|
|
let ctx = ParserContext {
|
|
four_octet_asn: true,
|
|
ibgp: false,
|
|
add_path: Vec::new(),
|
|
};
|
|
let mut wire = BytesMut::new();
|
|
Message::Update(update.clone())
|
|
.to_wire(&ctx, &mut wire)
|
|
.unwrap();
|
|
assert_eq!(
|
|
Message::decode(&ctx, &wire).unwrap(),
|
|
Message::Update(update)
|
|
);
|
|
|
|
let withdrawal = build_withdraw(&prefix("10.1.0.0/16"), true);
|
|
assert!(withdrawal.withdrawn_routes.is_empty());
|
|
assert_eq!(
|
|
withdrawal.path_attributes,
|
|
[PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute {
|
|
family: AfiSafi::IPV4_UNICAST,
|
|
prefixes: vec![prefix("10.1.0.0/16").into()],
|
|
})]
|
|
);
|
|
// An IPv6 route is as it always was, agreed or not.
|
|
let v6 = LocalRoute::new(prefix("2001:db8:1::/48"), LocalRouteAttrs::default());
|
|
assert_eq!(
|
|
build_announce(&v6, &agreed),
|
|
build_announce(&v6, &over_ipv6)
|
|
);
|
|
}
|
|
|
|
/// RFC 9234 §5, the egress procedure for routes we originate.
|
|
#[test]
|
|
fn a_route_is_marked_only_to_customer_on_its_way_down_or_across() {
|
|
let route = LocalRoute::new(
|
|
IpPrefix::try_from("10.1.0.0/16").unwrap(),
|
|
LocalRouteAttrs::default(),
|
|
);
|
|
let to = |role| Towards {
|
|
role,
|
|
..towards(65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), true)
|
|
};
|
|
let mark = |role| {
|
|
build_announce(&route, &to(role))
|
|
.unwrap()
|
|
.path_attributes
|
|
.into_iter()
|
|
.find(|attribute| matches!(attribute, PathAttribute::OnlyToCustomer(_)))
|
|
};
|
|
for role in [
|
|
PeerRole::Customer,
|
|
PeerRole::Peer,
|
|
PeerRole::RouteServerClient,
|
|
] {
|
|
assert_eq!(
|
|
mark(Some(role)),
|
|
Some(PathAttribute::OnlyToCustomer(65000)),
|
|
"{role:?}"
|
|
);
|
|
}
|
|
// On its way up it may still go anywhere; and where it is not
|
|
// known what the peer is, nothing is said.
|
|
for role in [Some(PeerRole::Provider), Some(PeerRole::RouteServer), None] {
|
|
assert_eq!(mark(role), None, "{role:?}");
|
|
}
|
|
}
|
|
|
|
/// An external peer.
|
|
fn towards(local_asn: u32, next_hop: IpAddr, four_octet_asn: bool) -> Towards {
|
|
Towards {
|
|
local_asn,
|
|
next_hop,
|
|
four_octet_asn,
|
|
ibgp: false,
|
|
role: None,
|
|
extended_next_hop: false,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn ipv4_announce_has_expected_attributes_and_roundtrips() {
|
|
let route = LocalRoute::new(
|
|
prefix("10.100.1.0/24"),
|
|
LocalRouteAttrs {
|
|
origin: OriginPathAttribute::IGP,
|
|
prepend: 2,
|
|
med: Some(50),
|
|
communities: vec![(65000, 100), (65535, 65281)],
|
|
large_communities: vec![LargeCommunity {
|
|
global_admin: 65000,
|
|
data_1: 1,
|
|
data_2: 2,
|
|
}],
|
|
},
|
|
);
|
|
let update = build_announce(
|
|
&route,
|
|
&towards(65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), true),
|
|
)
|
|
.unwrap();
|
|
|
|
assert_eq!(update.announced_routes, vec![prefix("10.100.1.0/24")]);
|
|
assert!(update.withdrawn_routes.is_empty());
|
|
assert_eq!(
|
|
update.path_attributes,
|
|
vec![
|
|
PathAttribute::Origin(OriginPathAttribute::IGP),
|
|
AsPathAttribute::from_asns(vec![65000, 65000, 65000]),
|
|
PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
|
|
PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(50)),
|
|
PathAttribute::Communities(CommunitiesPathAttribute(vec![
|
|
(65000, 100),
|
|
(65535, 65281)
|
|
])),
|
|
PathAttribute::LargeCommunities(LargeCommunitiesPathAttribute {
|
|
communities: vec![LargeCommunity {
|
|
global_admin: 65000,
|
|
data_1: 1,
|
|
data_2: 2
|
|
}],
|
|
}),
|
|
]
|
|
);
|
|
|
|
// Serialise and parse back to make sure the wire encoding is sound.
|
|
let ctx = ParserContext {
|
|
four_octet_asn: true,
|
|
ibgp: false,
|
|
add_path: Vec::new(),
|
|
};
|
|
let msg = Message::Update(update.clone());
|
|
let mut buf = BytesMut::new();
|
|
msg.to_wire(&ctx, &mut buf).unwrap();
|
|
let Message::Update(parsed) = Message::decode(&ctx, &buf).unwrap() else {
|
|
panic!("expected UPDATE")
|
|
};
|
|
assert_eq!(parsed, update);
|
|
}
|
|
|
|
#[test]
|
|
fn ipv6_announce_uses_mp_reach() {
|
|
let route = LocalRoute::new(prefix("2001:db8:1::/48"), LocalRouteAttrs::default());
|
|
let nh = Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1);
|
|
let update = build_announce(&route, &towards(65000, IpAddr::V6(nh), true)).unwrap();
|
|
|
|
assert!(update.announced_routes.is_empty());
|
|
assert!(
|
|
!update
|
|
.path_attributes
|
|
.iter()
|
|
.any(|a| matches!(a, PathAttribute::NextHop(_)))
|
|
);
|
|
assert!(update.path_attributes.iter().any(|a| matches!(
|
|
a,
|
|
PathAttribute::MpReachNlri(MpReachNlriPathAttribute { next_hop: NlriNextHop::Ipv6(h), prefixes, .. })
|
|
if *h == nh && *prefixes == vec![prefix("2001:db8:1::/48")]
|
|
)));
|
|
}
|
|
|
|
/// A two-octet session cannot carry a four-octet AS number in AS_PATH:
|
|
/// it goes out as AS_TRANS there, and in full in AS4_PATH.
|
|
#[test]
|
|
fn a_wide_as_number_reaches_a_two_octet_peer_in_as4_path() {
|
|
let route = LocalRoute::new(
|
|
prefix("10.100.1.0/24"),
|
|
LocalRouteAttrs {
|
|
prepend: 1,
|
|
..LocalRouteAttrs::default()
|
|
},
|
|
);
|
|
let next_hop = IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1));
|
|
let path = |update: &UpdateMessage, as4: bool| {
|
|
update.path_attributes.iter().find_map(|a| match a {
|
|
PathAttribute::As4Path(p) if as4 => Some(p.segments[0].path.clone()),
|
|
PathAttribute::AsPath(p) if !as4 => Some(p.segments[0].path.clone()),
|
|
_ => None,
|
|
})
|
|
};
|
|
|
|
let update = build_announce(&route, &towards(4_200_000_001, next_hop, false)).unwrap();
|
|
assert_eq!(path(&update, true), Some(vec![4_200_000_001; 2]));
|
|
|
|
// What the peer reads from the wire.
|
|
let ctx = ParserContext {
|
|
four_octet_asn: false,
|
|
ibgp: false,
|
|
add_path: Vec::new(),
|
|
};
|
|
let mut buf = BytesMut::new();
|
|
Message::Update(update).to_wire(&ctx, &mut buf).unwrap();
|
|
let Message::Update(seen) = Message::decode(&ctx, &buf).unwrap() else {
|
|
panic!("expected UPDATE")
|
|
};
|
|
assert_eq!(path(&seen, false), Some(vec![23456; 2]), "AS_TRANS");
|
|
assert_eq!(path(&seen, true), Some(vec![4_200_000_001; 2]));
|
|
|
|
// No AS4_PATH when it would say nothing new: the peer handles
|
|
// four-octet numbers, or ours fits in two.
|
|
let update = build_announce(&route, &towards(4_200_000_001, next_hop, true)).unwrap();
|
|
assert_eq!(path(&update, true), None);
|
|
let update = build_announce(&route, &towards(65000, next_hop, false)).unwrap();
|
|
assert_eq!(path(&update, true), None);
|
|
assert_eq!(path(&update, false), Some(vec![65000; 2]));
|
|
}
|
|
|
|
#[test]
|
|
fn family_mismatch_is_skipped() {
|
|
let v6 = LocalRoute::new(prefix("2001:db8::/32"), LocalRouteAttrs::default());
|
|
assert!(
|
|
build_announce(&v6, &towards(65000, IpAddr::V4(Ipv4Addr::LOCALHOST), true)).is_none()
|
|
);
|
|
let v4 = LocalRoute::new(prefix("10.0.0.0/8"), LocalRouteAttrs::default());
|
|
assert!(
|
|
build_announce(&v4, &towards(65000, IpAddr::V6(Ipv6Addr::LOCALHOST), true)).is_none()
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn withdraw_by_family() {
|
|
let v4 = build_withdraw(&prefix("10.0.0.0/8"), false);
|
|
assert_eq!(v4.withdrawn_routes, vec![prefix("10.0.0.0/8")]);
|
|
assert!(v4.path_attributes.is_empty());
|
|
|
|
let v6 = build_withdraw(&prefix("2001:db8::/32"), false);
|
|
assert!(v6.withdrawn_routes.is_empty());
|
|
assert!(
|
|
matches!(&v6.path_attributes[..], [PathAttribute::MpUnreachNlri(m)] if m.prefixes == vec![prefix("2001:db8::/32")])
|
|
);
|
|
}
|
|
|
|
/// Inside our own AS a route we originate has an empty AS path and a
|
|
/// LOCAL_PREF. With our AS number on the path, a peer in our AS takes
|
|
/// the route for a loop; without LOCAL_PREF it treats the route as
|
|
/// withdrawn, as our own decoder does.
|
|
#[test]
|
|
fn a_route_announced_inside_our_as_has_an_empty_path_and_a_local_pref() {
|
|
let route = LocalRoute::new(
|
|
prefix("10.100.1.0/24"),
|
|
LocalRouteAttrs {
|
|
prepend: 3,
|
|
..LocalRouteAttrs::default()
|
|
},
|
|
);
|
|
let internal = Towards {
|
|
ibgp: true,
|
|
..towards(65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), true)
|
|
};
|
|
let update = build_announce(&route, &internal).unwrap();
|
|
assert_eq!(
|
|
update.path_attributes,
|
|
vec![
|
|
PathAttribute::Origin(OriginPathAttribute::IGP),
|
|
PathAttribute::AsPath(AsPathAttribute { segments: vec![] }),
|
|
PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
|
|
PathAttribute::LocalPref(LocalPrefPathAttribute(100)),
|
|
]
|
|
);
|
|
|
|
// An internal peer accepts it as it stands.
|
|
let ctx = ParserContext {
|
|
four_octet_asn: true,
|
|
ibgp: true,
|
|
add_path: Vec::new(),
|
|
};
|
|
let mut buf = BytesMut::new();
|
|
Message::Update(update.clone())
|
|
.to_wire(&ctx, &mut buf)
|
|
.unwrap();
|
|
let decoded = Message::decode_with_report(&ctx, &buf).unwrap();
|
|
assert!(
|
|
decoded.update_errors.is_empty(),
|
|
"{:?}",
|
|
decoded.update_errors
|
|
);
|
|
assert_eq!(decoded.message, Message::Update(update));
|
|
|
|
// To an external peer the same route has our AS on the path, as
|
|
// many times as configured, and no LOCAL_PREF.
|
|
let external = Towards {
|
|
ibgp: false,
|
|
..internal
|
|
};
|
|
let update = build_announce(&route, &external).unwrap();
|
|
assert_eq!(
|
|
update.path_attributes[1],
|
|
AsPathAttribute::from_asns(vec![65000; 4])
|
|
);
|
|
assert!(
|
|
!update
|
|
.path_attributes
|
|
.iter()
|
|
.any(|a| matches!(a, PathAttribute::LocalPref(_)))
|
|
);
|
|
}
|
|
|
|
// -- Exporters --
|
|
|
|
use std::time::Duration;
|
|
|
|
use bgp_rib::table::{PeerInfo, RibTable};
|
|
use tokio::sync::mpsc;
|
|
|
|
use crate::collision::Direction;
|
|
use crate::family::FamilySet;
|
|
use crate::fsm::Negotiated;
|
|
use crate::session::SessionGuard;
|
|
|
|
/// A registry and local RIB with an advertiser running, and a table to
|
|
/// take peer numbers from.
|
|
struct Fixture {
|
|
registry: SessionRegistry,
|
|
local_rib: LocalRib,
|
|
table: RibTable,
|
|
}
|
|
|
|
/// The far end of a session's queue: what the session's task would
|
|
/// write to the peer.
|
|
struct Queue {
|
|
outbound: mpsc::Receiver<Message>,
|
|
/// What the peer holds, going by what it has been sent.
|
|
routes: HashSet<IpPrefix>,
|
|
_registration: SessionGuard,
|
|
}
|
|
|
|
impl Fixture {
|
|
fn new() -> Self {
|
|
let registry = SessionRegistry::new();
|
|
let local_rib = LocalRib::spawn();
|
|
spawn_advertiser(
|
|
local_rib.clone(),
|
|
Arc::new(RibTable::new(4)),
|
|
registry.clone(),
|
|
65000,
|
|
None,
|
|
Ipv4Addr::new(192, 0, 2, 1),
|
|
watch::channel(true).1,
|
|
);
|
|
Self {
|
|
registry,
|
|
local_rib,
|
|
table: RibTable::new(1),
|
|
}
|
|
}
|
|
|
|
/// A session with peer number `n`, whose queue holds `capacity`
|
|
/// messages.
|
|
fn session(&self, n: u8, capacity: usize) -> Queue {
|
|
let address = IpAddr::V4(Ipv4Addr::new(192, 0, 2, n));
|
|
let negotiated = Negotiated {
|
|
peer_asn: 65000 + u32::from(n),
|
|
ibgp: false,
|
|
peer_identifier: Ipv4Addr::new(10, 0, 0, n),
|
|
hold_time_secs: 90,
|
|
four_octet_asn: true,
|
|
route_refresh: false,
|
|
families: FamilySet::supported(),
|
|
add_path_receive: FamilySet::default(),
|
|
add_path_send: FamilySet::default(),
|
|
role_confirmed: false,
|
|
extended_next_hop: false,
|
|
graceful_restart: None,
|
|
};
|
|
let rib_peer = self
|
|
.table
|
|
.add_peer(PeerInfo {
|
|
asn: negotiated.peer_asn,
|
|
router_id: negotiated.peer_identifier,
|
|
address,
|
|
interface: 0,
|
|
ibgp: false,
|
|
reflector_client: false,
|
|
})
|
|
.unwrap();
|
|
let (tx, outbound) = mpsc::channel(capacity);
|
|
let handle = SessionHandle::new(
|
|
PeerId {
|
|
addr: (address, 179).into(),
|
|
asn: negotiated.peer_asn,
|
|
},
|
|
None,
|
|
Direction::Inbound,
|
|
(Ipv4Addr::new(192, 0, 2, 100), 179).into(),
|
|
&negotiated,
|
|
rib_peer,
|
|
tx,
|
|
);
|
|
Queue {
|
|
outbound,
|
|
routes: HashSet::new(),
|
|
_registration: self.registry.register(handle),
|
|
}
|
|
}
|
|
|
|
async fn announce(&self, n: u32) {
|
|
let route = LocalRoute::new(nth_prefix(n), LocalRouteAttrs::default());
|
|
self.local_rib.announce(route).await.unwrap();
|
|
}
|
|
|
|
async fn routes(&self) -> HashSet<IpPrefix> {
|
|
let routes = self.local_rib.snapshot().await.unwrap();
|
|
routes.iter().map(|route| route.prefix).collect()
|
|
}
|
|
}
|
|
|
|
fn nth_prefix(n: u32) -> IpPrefix {
|
|
let bytes = (0x0a00_0000 + (n << 8)).to_be_bytes();
|
|
IpPrefix::new(IpVersion::Ipv4, &bytes[..3], 24).unwrap()
|
|
}
|
|
|
|
impl Queue {
|
|
/// Reads what has been queued for the peer until it holds exactly
|
|
/// `expected`, as a peer would that takes every message in turn.
|
|
async fn read_until_it_has(&mut self, expected: &HashSet<IpPrefix>) {
|
|
let reading = async {
|
|
while self.routes != *expected {
|
|
let Some(Message::Update(update)) = self.outbound.recv().await else {
|
|
panic!("expected an UPDATE");
|
|
};
|
|
for nlri in &update.withdrawn_routes {
|
|
self.routes.remove(&nlri.prefix);
|
|
}
|
|
for nlri in &update.announced_routes {
|
|
self.routes.insert(nlri.prefix);
|
|
}
|
|
}
|
|
};
|
|
if tokio::time::timeout(Duration::from_secs(10), reading)
|
|
.await
|
|
.is_err()
|
|
{
|
|
panic!(
|
|
"the peer has {} routes, {} of them right, and should have {}",
|
|
self.routes.len(),
|
|
self.routes.intersection(expected).count(),
|
|
expected.len()
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A route from a router of our AS reaches a route reflector client
|
|
/// of ours through the advertiser, marked with where it started and our
|
|
/// cluster, with its preference, and with us as next hop.
|
|
#[tokio::test]
|
|
async fn a_route_from_one_of_our_routers_is_reflected_to_a_client() {
|
|
let registry = SessionRegistry::new();
|
|
let rib = Arc::new(RibTable::new(4));
|
|
let cluster = Ipv4Addr::new(10, 0, 0, 100);
|
|
spawn_advertiser(
|
|
LocalRib::spawn(),
|
|
Arc::clone(&rib),
|
|
registry.clone(),
|
|
65000,
|
|
None,
|
|
cluster,
|
|
watch::channel(true).1,
|
|
);
|
|
let ours = |n: u8, client| PeerInfo {
|
|
asn: 65000,
|
|
router_id: Ipv4Addr::new(10, 0, 0, n),
|
|
address: IpAddr::V4(Ipv4Addr::new(192, 0, 2, n)),
|
|
interface: 0,
|
|
ibgp: true,
|
|
reflector_client: client,
|
|
};
|
|
// Another router of ours, not a client, with a route in the table.
|
|
let other = rib.add_peer(ours(1, false)).unwrap();
|
|
let came = [
|
|
PathAttribute::Origin(OriginPathAttribute::IGP),
|
|
AsPathAttribute::from_asns(vec![64496]),
|
|
PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
|
|
PathAttribute::LocalPref(LocalPrefPathAttribute(200)),
|
|
];
|
|
let attrs = rib.attrs().intern(&came, None).unwrap();
|
|
rib.announce(other, &attrs, &[Nlri::from(nth_prefix(1))]);
|
|
|
|
// A client that is told of learned routes.
|
|
let negotiated = Negotiated {
|
|
peer_asn: 65000,
|
|
ibgp: true,
|
|
peer_identifier: Ipv4Addr::new(10, 0, 0, 2),
|
|
hold_time_secs: 90,
|
|
four_octet_asn: true,
|
|
route_refresh: false,
|
|
families: FamilySet::supported(),
|
|
add_path_receive: FamilySet::default(),
|
|
add_path_send: FamilySet::default(),
|
|
role_confirmed: false,
|
|
extended_next_hop: false,
|
|
graceful_restart: None,
|
|
};
|
|
let rib_peer = rib.add_peer(ours(2, true)).unwrap();
|
|
let (tx, mut outbound) = mpsc::channel(16);
|
|
let mut handle = SessionHandle::new(
|
|
PeerId {
|
|
addr: (Ipv4Addr::new(192, 0, 2, 2), 179).into(),
|
|
asn: 65000,
|
|
},
|
|
None,
|
|
Direction::Inbound,
|
|
(Ipv4Addr::new(192, 0, 2, 100), 179).into(),
|
|
&negotiated,
|
|
rib_peer,
|
|
tx,
|
|
);
|
|
handle.export = Arc::new(crate::policy::ExportPolicy {
|
|
learned_routes: true,
|
|
..crate::policy::ExportPolicy::default()
|
|
});
|
|
handle.reflector_client = true;
|
|
let _registration = registry.register(handle);
|
|
|
|
let update = tokio::time::timeout(Duration::from_secs(10), async {
|
|
loop {
|
|
match outbound.recv().await {
|
|
Some(Message::Update(update)) if !update.announced_routes.is_empty() => {
|
|
return update;
|
|
}
|
|
Some(_) => {}
|
|
None => panic!("the session's queue closed"),
|
|
}
|
|
}
|
|
})
|
|
.await
|
|
.expect("the client is told of the route");
|
|
assert_eq!(update.announced_routes, [Nlri::from(nth_prefix(1))]);
|
|
assert_eq!(
|
|
update.path_attributes,
|
|
[
|
|
PathAttribute::Origin(OriginPathAttribute::IGP),
|
|
AsPathAttribute::from_asns(vec![64496]),
|
|
PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 100))),
|
|
PathAttribute::LocalPref(LocalPrefPathAttribute(200)),
|
|
PathAttribute::OriginatorId(Ipv4Addr::new(10, 0, 0, 1)),
|
|
PathAttribute::ClusterList(vec![cluster]),
|
|
]
|
|
);
|
|
}
|
|
|
|
/// One peer that takes nothing must not keep the others from being
|
|
/// sent their routes. There used to be one task announcing to every
|
|
/// session in turn, and it waited for whichever queue was full.
|
|
#[tokio::test]
|
|
async fn a_peer_that_reads_nothing_holds_up_nobody_else() {
|
|
let fixture = Fixture::new();
|
|
let _stuck = fixture.session(1, 4);
|
|
let mut reading = fixture.session(2, 4);
|
|
|
|
for n in 0..500 {
|
|
fixture.announce(n).await;
|
|
}
|
|
reading.read_until_it_has(&fixture.routes().await).await;
|
|
assert_eq!(reading.routes.len(), 500);
|
|
}
|
|
|
|
/// A peer that reads so slowly that changes pass it by is brought up
|
|
/// to date when it reads again: with the routes there now are, and
|
|
/// without the ones that went while it was not looking. Those used to
|
|
/// stay announced to it for good.
|
|
#[tokio::test]
|
|
async fn a_peer_that_fell_behind_ends_up_with_the_routes_there_are() {
|
|
let fixture = Fixture::new();
|
|
let mut slow = fixture.session(1, 4);
|
|
for n in 0..10 {
|
|
fixture.announce(n).await;
|
|
}
|
|
slow.read_until_it_has(&fixture.routes().await).await;
|
|
|
|
// Far more changes than are kept for a subscriber that is not
|
|
// reading, among them the withdrawal of what the peer has.
|
|
for n in 0..10 {
|
|
fixture.local_rib.withdraw(nth_prefix(n)).await.unwrap();
|
|
}
|
|
for n in 10..3000 {
|
|
fixture.announce(n).await;
|
|
}
|
|
for n in 2000..3000 {
|
|
fixture.local_rib.withdraw(nth_prefix(n)).await.unwrap();
|
|
}
|
|
|
|
let routes = fixture.routes().await;
|
|
assert_eq!(routes.len(), 1990);
|
|
slow.read_until_it_has(&routes).await;
|
|
}
|
|
|
|
/// When a peer's session is replaced by a new one, the new session is
|
|
/// sent the routes from the beginning.
|
|
#[tokio::test]
|
|
async fn a_new_session_with_the_same_peer_starts_afresh() {
|
|
let fixture = Fixture::new();
|
|
fixture.announce(1).await;
|
|
let routes = fixture.routes().await;
|
|
|
|
let mut first = fixture.session(1, 4);
|
|
first.read_until_it_has(&routes).await;
|
|
drop(first);
|
|
let mut second = fixture.session(1, 4);
|
|
second.read_until_it_has(&routes).await;
|
|
}
|
|
|
|
// -- Export limits --
|
|
|
|
use crate::policy::{ExportLimits, ExportPolicy, PrefixSet};
|
|
|
|
/// A table with routes from a customer for `prefixes`, an advertiser
|
|
/// over it with `originated` routes of our own, and a session to a
|
|
/// peer of ours that is told of learned routes, within `limits`.
|
|
async fn limited(
|
|
prefixes: &[IpPrefix],
|
|
originated: &[IpPrefix],
|
|
limits: ExportLimits,
|
|
) -> LimitedPeer {
|
|
let registry = SessionRegistry::new();
|
|
let rib = Arc::new(RibTable::new(4));
|
|
let local_rib = LocalRib::spawn();
|
|
for prefix in originated {
|
|
let route = LocalRoute::new(*prefix, LocalRouteAttrs::default());
|
|
local_rib.announce(route).await.unwrap();
|
|
}
|
|
spawn_advertiser(
|
|
local_rib,
|
|
Arc::clone(&rib),
|
|
registry.clone(),
|
|
65000,
|
|
None,
|
|
Ipv4Addr::new(192, 0, 2, 1),
|
|
watch::channel(true).1,
|
|
);
|
|
let peer = |n: u8, asn| PeerInfo {
|
|
asn,
|
|
router_id: Ipv4Addr::new(10, 0, 0, n),
|
|
address: IpAddr::V4(Ipv4Addr::new(192, 0, 2, n)),
|
|
interface: 0,
|
|
ibgp: false,
|
|
reflector_client: false,
|
|
};
|
|
// A customer's routes carry no Only to Customer mark: as far as
|
|
// the roles go, they are for everyone.
|
|
let customer = rib.add_peer(peer(1, 64496)).unwrap();
|
|
let came = [
|
|
PathAttribute::Origin(OriginPathAttribute::IGP),
|
|
AsPathAttribute::from_asns(vec![64496]),
|
|
PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
|
|
];
|
|
let attrs = rib.attrs().intern(&came, None).unwrap();
|
|
let nlri: Vec<Nlri> = prefixes.iter().copied().map(Nlri::from).collect();
|
|
rib.announce(customer, &attrs, &nlri);
|
|
|
|
let negotiated = Negotiated {
|
|
peer_asn: 64497,
|
|
ibgp: false,
|
|
peer_identifier: Ipv4Addr::new(10, 0, 0, 2),
|
|
hold_time_secs: 90,
|
|
four_octet_asn: true,
|
|
route_refresh: false,
|
|
families: FamilySet::supported(),
|
|
add_path_receive: FamilySet::default(),
|
|
add_path_send: FamilySet::default(),
|
|
role_confirmed: false,
|
|
extended_next_hop: false,
|
|
graceful_restart: None,
|
|
};
|
|
let rib_peer = rib.add_peer(peer(2, 64497)).unwrap();
|
|
let (tx, outbound) = mpsc::channel(1024);
|
|
let mut handle = SessionHandle::new(
|
|
PeerId {
|
|
addr: (Ipv4Addr::new(192, 0, 2, 2), 179).into(),
|
|
asn: 64497,
|
|
},
|
|
None,
|
|
Direction::Inbound,
|
|
(Ipv4Addr::new(192, 0, 2, 100), 179).into(),
|
|
&negotiated,
|
|
rib_peer,
|
|
tx,
|
|
);
|
|
handle.role = Some(PeerRole::Peer);
|
|
handle.export = Arc::new(ExportPolicy {
|
|
learned_routes: true,
|
|
..ExportPolicy::default()
|
|
});
|
|
handle.export_limits = Some(Arc::new(limits));
|
|
let ending = handle.listen_for_end();
|
|
LimitedPeer {
|
|
outbound,
|
|
ending,
|
|
_registration: registry.register(handle),
|
|
}
|
|
}
|
|
|
|
/// The far end of the limited peer's session: what it is sent, and
|
|
/// what ends it.
|
|
struct LimitedPeer {
|
|
outbound: mpsc::Receiver<Message>,
|
|
ending: mpsc::UnboundedReceiver<SessionError>,
|
|
_registration: SessionGuard,
|
|
}
|
|
|
|
/// The routes announced on a session's queue until it has been quiet
|
|
/// for a while.
|
|
async fn announced(outbound: &mut mpsc::Receiver<Message>) -> HashSet<IpPrefix> {
|
|
let mut routes = HashSet::new();
|
|
while let Ok(Some(message)) =
|
|
tokio::time::timeout(Duration::from_millis(500), outbound.recv()).await
|
|
{
|
|
if let Message::Update(update) = message {
|
|
for nlri in &update.withdrawn_routes {
|
|
routes.remove(&nlri.prefix);
|
|
}
|
|
routes.extend(update.announced_routes.iter().map(|nlri| nlri.prefix));
|
|
if let Some(PathAttribute::MpReachNlri(reach)) = update
|
|
.path_attributes
|
|
.iter()
|
|
.find(|a| matches!(a, PathAttribute::MpReachNlri(_)))
|
|
{
|
|
routes.extend(reach.prefixes.iter().map(|nlri| nlri.prefix));
|
|
}
|
|
}
|
|
}
|
|
routes
|
|
}
|
|
|
|
fn allowing(prefixes: &[IpPrefix]) -> Option<PrefixSet> {
|
|
let mut set = PrefixSet::default();
|
|
for prefix in prefixes {
|
|
set.allow(*prefix, prefix.length());
|
|
}
|
|
Some(set)
|
|
}
|
|
|
|
/// Only routes for the prefixes in the export limits go to a peer,
|
|
/// learned or our own, though nothing else keeps the others from it:
|
|
/// a customer's routes are for everyone, as far as the roles go.
|
|
#[tokio::test]
|
|
async fn routes_outside_the_export_limits_do_not_go_to_a_peer() {
|
|
let limits = ExportLimits {
|
|
prefixes: allowing(&[nth_prefix(2), nth_prefix(11)]),
|
|
max_prefixes: None,
|
|
};
|
|
let mut peer = limited(
|
|
&[nth_prefix(1), nth_prefix(2), nth_prefix(3)],
|
|
&[nth_prefix(10), nth_prefix(11)],
|
|
limits,
|
|
)
|
|
.await;
|
|
assert_eq!(
|
|
announced(&mut peer.outbound).await,
|
|
HashSet::from([nth_prefix(2), nth_prefix(11)])
|
|
);
|
|
assert!(peer.ending.try_recv().is_err(), "the session goes on");
|
|
}
|
|
|
|
/// When more routes would go to a peer than its export limit allows,
|
|
/// none of them does, and the session is ended: a leak becomes a
|
|
/// session that is down, and not a table spread across the Internet.
|
|
#[tokio::test]
|
|
async fn more_routes_than_the_export_limit_end_the_session_and_none_goes() {
|
|
let prefixes: Vec<IpPrefix> = (1..=5).map(nth_prefix).collect();
|
|
let limits = ExportLimits {
|
|
prefixes: None,
|
|
max_prefixes: Some(3),
|
|
};
|
|
let mut peer = limited(&prefixes, &[], limits).await;
|
|
let why = tokio::time::timeout(Duration::from_secs(10), peer.ending.recv())
|
|
.await
|
|
.expect("the session is asked to end")
|
|
.expect("by the exporter");
|
|
assert!(
|
|
matches!(
|
|
why,
|
|
SessionError::ExportLimitExceeded {
|
|
limit: 3,
|
|
routes: 5
|
|
}
|
|
),
|
|
"{why:?}"
|
|
);
|
|
assert_eq!(announced(&mut peer.outbound).await, HashSet::new());
|
|
}
|
|
|
|
/// Our own routes count too, and none goes past the limit: those that
|
|
/// fit go, and the one that does not ends the session.
|
|
#[tokio::test]
|
|
async fn originated_routes_count_towards_the_export_limit() {
|
|
let originated: Vec<IpPrefix> = (10..13).map(nth_prefix).collect();
|
|
let limits = ExportLimits {
|
|
prefixes: None,
|
|
max_prefixes: Some(2),
|
|
};
|
|
let mut peer = limited(&[], &originated, limits).await;
|
|
let why = tokio::time::timeout(Duration::from_secs(10), peer.ending.recv())
|
|
.await
|
|
.expect("the session is asked to end")
|
|
.expect("by the exporter");
|
|
assert!(
|
|
matches!(
|
|
why,
|
|
SessionError::ExportLimitExceeded {
|
|
limit: 2,
|
|
routes: 3
|
|
}
|
|
),
|
|
"{why:?}"
|
|
);
|
|
assert_eq!(announced(&mut peer.outbound).await.len(), 2);
|
|
}
|
|
}
|