From ad072331967a0f5af6b83348e06b1379933ad8bd Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Wed, 16 Sep 2026 22:56:29 +0200 Subject: [PATCH] feature: locally originated routes, session registry, outbound UPDATEs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 of the BGP teaching demo: the daemon can now originate routes and advertise them to every established peer. - rib: add LocalRib actor holding locally originated routes with owner, origin, AS-path prepend, MED, standard and large communities. Enforces a LocalRibPolicy (prefix allowlist, per-owner limit, owner match on replace/withdraw) and emits LocalRibEvents. - server: add SessionRegistry with per-session outbound queues; sessions register on Established and deregister via a drop guard. Emits SessionEvent::{Up,Down}. - server: advertiser task fans LocalRib changes out to all sessions and replays the table to new sessions. UPDATE builders handle IPv4 (legacy NLRI) and IPv6 (MP_REACH/MP_UNREACH). - fsm: merge the timer and timer-less established loops into one select! with an outbound branch; reset the hold timer on any received message per RFC 4271 §8.2.2 (previously only on KEEPALIVE). - config: new local_route_policy proto message. - packet: fix Communitites typo, expose CommunitiesPathAttribute inner vec. - primitives: IpPrefix::contains. - integration tests: announce/withdraw/replay/fan-out/policy coverage. Co-Authored-By: Claude Fable 5.1 --- Cargo.lock | 3 + crates/integration-tests/src/test_peer.rs | 14 +- crates/integration-tests/src/test_server.rs | 21 +- crates/integration-tests/tests/announce.rs | 256 ++++++++++++++++ crates/packet/src/message.rs | 22 +- crates/primitives/src/ip_prefix.rs | 35 +++ crates/rib/Cargo.toml | 2 + crates/rib/src/lib.rs | 2 + crates/rib/src/local_rib.rs | 305 ++++++++++++++++++++ crates/server/Cargo.toml | 1 + crates/server/proto/config.proto | 14 + crates/server/src/advertise.rs | 267 +++++++++++++++++ crates/server/src/config.rs | 27 +- crates/server/src/fsm.rs | 94 +++--- crates/server/src/lib.rs | 2 + crates/server/src/server.rs | 19 +- crates/server/src/session.rs | 118 ++++++++ docs/RFCs.md | 7 +- testdata/server.textpb | 5 + 19 files changed, 1144 insertions(+), 70 deletions(-) create mode 100644 crates/integration-tests/tests/announce.rs create mode 100644 crates/rib/src/local_rib.rs create mode 100644 crates/server/src/advertise.rs create mode 100644 crates/server/src/session.rs diff --git a/Cargo.lock b/Cargo.lock index bb01e18..9b0d826 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -159,6 +159,8 @@ version = "0.0.0" dependencies = [ "bgp-packet", "bgp-primitives", + "serde", + "thiserror 2.0.12", "tokio", "tracing", ] @@ -177,6 +179,7 @@ dependencies = [ "prometheus", "protobuf", "protobuf-codegen", + "thiserror 2.0.12", "tokio", "tracing", "tracing-subscriber", diff --git a/crates/integration-tests/src/test_peer.rs b/crates/integration-tests/src/test_peer.rs index fc54ba7..380763b 100644 --- a/crates/integration-tests/src/test_peer.rs +++ b/crates/integration-tests/src/test_peer.rs @@ -1,6 +1,6 @@ use std::net::{Ipv4Addr, SocketAddr}; -use bgp_packet::message::{BGP_HEADER_LEN, Message}; +use bgp_packet::message::{BGP_HEADER_LEN, Message, UpdateMessage}; use bgp_packet::open::{Capability, OpenMessage, OpenOption}; use bgp_primitives::constants::{AddressFamilyId, SubsequentAfi}; use bgp_primitives::parser::ParserContext; @@ -49,6 +49,18 @@ impl TestPeer { Ok(msg) } + /// Read messages until an UPDATE arrives, skipping KEEPALIVEs. Any other + /// message type is an error. + pub async fn recv_update(&mut self) -> Result { + loop { + match self.recv().await? { + Message::Update(u) => return Ok(u), + Message::KeepAlive => {} + other => bail!("Expected UPDATE, got {:?}", other), + } + } + } + /// Receive and unwrap as `OpenMessage`, failing if a different type arrives. pub async fn recv_open(&mut self) -> Result { match self.recv().await? { diff --git a/crates/integration-tests/src/test_server.rs b/crates/integration-tests/src/test_server.rs index d41fa4d..139db1d 100644 --- a/crates/integration-tests/src/test_server.rs +++ b/crates/integration-tests/src/test_server.rs @@ -1,8 +1,9 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr}; -use bgp_rib::AdjRibIn; +use bgp_rib::{AdjRibIn, LocalRib, LocalRibPolicy}; use bgp_server::config::Config; use bgp_server::server::BgpServerBuilder; +use bgp_server::session::SessionRegistry; use eyre::Result; use tokio::task::JoinHandle; @@ -14,6 +15,10 @@ pub struct TestServer { pub addr: SocketAddr, /// Handle to the server's Adj-RIB-In for test introspection. pub adj_rib_in: AdjRibIn, + /// Handle to the server's locally originated routes. + pub local_rib: LocalRib, + /// Live view of the server's established sessions. + pub registry: SessionRegistry, } impl TestServer { @@ -24,16 +29,22 @@ impl TestServer { /// Start a server with an explicit hold time (useful for timer tests). pub async fn start_with_hold_time(local_asn: u32, hold_time_secs: u16) -> Result { - Self::start_inner(local_asn, hold_time_secs).await + Self::start_inner(local_asn, hold_time_secs, LocalRibPolicy::default()).await } - async fn start_inner(local_asn: u32, hold_time_secs: u16) -> Result { + /// Start a server with a policy restricting locally originated routes. + pub async fn start_with_policy(local_asn: u32, policy: LocalRibPolicy) -> Result { + Self::start_inner(local_asn, 90, policy).await + } + + async fn start_inner(local_asn: u32, hold_time_secs: u16, policy: LocalRibPolicy) -> Result { let config = Config { router_id: Ipv4Addr::new(127, 0, 0, 1), local_asn, hold_time_secs, listen_endpoints: vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)], peers: vec![], + local_route_policy: policy, }; let bound = BgpServerBuilder::new(config).bind().await?; @@ -43,9 +54,11 @@ impl TestServer { .next() .expect("at least one listener"); let adj_rib_in = bound.adj_rib_in.clone(); + let local_rib = bound.local_rib.clone(); + let registry = bound.registry.clone(); let handle = tokio::spawn(bound.run()); - Ok(Self { handle, addr, adj_rib_in }) + Ok(Self { handle, addr, adj_rib_in, local_rib, registry }) } } diff --git a/crates/integration-tests/tests/announce.rs b/crates/integration-tests/tests/announce.rs new file mode 100644 index 0000000..eca0aea --- /dev/null +++ b/crates/integration-tests/tests/announce.rs @@ -0,0 +1,256 @@ +//! End-to-end tests for locally originated routes: announcing via the +//! `LocalRib` handle must produce UPDATEs on every established session. + +use std::net::Ipv4Addr; +use std::time::Duration; + +use bgp_integration_tests::test_peer::TestPeer; +use bgp_integration_tests::test_server::TestServer; +use bgp_packet::message::{ + AsPathAttribute, CommunitiesPathAttribute, LargeCommunitiesPathAttribute, LargeCommunity, + NextHopPathAttribute, OriginPathAttribute, PathAttribute, UpdateMessage, +}; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rib::{LocalRibError, LocalRibPolicy, LocalRoute, LocalRouteAttrs}; +use bgp_server::session::SessionEvent; +use eyre::Result; + +const LOCAL_ASN: u32 = 65000; +const PEER_ASN: u32 = 65001; +const TIMEOUT: Duration = Duration::from_secs(5); + +fn prefix(s: &str) -> IpPrefix { + IpPrefix::try_from(s).unwrap() +} + +fn route(p: &str, attrs: LocalRouteAttrs) -> LocalRoute { + LocalRoute::new(prefix(p), "alice", attrs) +} + +async fn established_peer(server: &TestServer) -> Result { + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 1), 90).await?; + Ok(peer) +} + +async fn next_update(peer: &mut TestPeer) -> Result { + tokio::time::timeout(TIMEOUT, peer.recv_update()).await? +} + +/// Waits until the registry reports `n` established sessions. +async fn wait_for_sessions(server: &TestServer, n: usize) -> Result<()> { + tokio::time::timeout(TIMEOUT, async { + while server.registry.list().len() != n { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await?; + Ok(()) +} + +// --------------------------------------------------------------------------- +// Session registry +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn test_registry_tracks_session_lifecycle() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + let mut events = server.registry.subscribe(); + assert!(server.registry.list().is_empty()); + + let peer = established_peer(&server).await?; + + let up = tokio::time::timeout(TIMEOUT, events.recv()).await??; + let SessionEvent::Up(handle) = up else { panic!("expected Up, got {up:?}") }; + assert_eq!(handle.peer.asn, PEER_ASN); + assert_eq!(handle.local_addr, server.addr); + assert!(handle.four_octet_asn); + assert_eq!(server.registry.list().len(), 1); + + drop(peer); + + let down = tokio::time::timeout(TIMEOUT, events.recv()).await??; + assert!(matches!(down, SessionEvent::Down(p) if p == handle.peer), "got {down:?}"); + assert!(server.registry.list().is_empty()); + Ok(()) +} + +// --------------------------------------------------------------------------- +// Announce / withdraw +// --------------------------------------------------------------------------- + +/// A route announced while the session is up is sent immediately, with the +/// attributes the announcer chose and our address as the next hop. +#[tokio::test] +async fn test_announce_after_session_up() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + let mut peer = established_peer(&server).await?; + wait_for_sessions(&server, 1).await?; + + server + .local_rib + .announce(route( + "10.100.1.0/24", + LocalRouteAttrs { + origin: OriginPathAttribute::IGP, + prepend: 0, + med: None, + communities: vec![(65000, 100), (65535, 65281)], + large_communities: vec![LargeCommunity { global_admin: 65000, data_1: 7, data_2: 8 }], + }, + )) + .await?; + + let update = next_update(&mut peer).await?; + 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![LOCAL_ASN]), + PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::LOCALHOST)), + PathAttribute::Communities(CommunitiesPathAttribute(vec![(65000, 100), (65535, 65281)])), + PathAttribute::LargeCommunities(LargeCommunitiesPathAttribute { + communities: vec![LargeCommunity { global_admin: 65000, data_1: 7, data_2: 8 }], + }), + ] + ); + Ok(()) +} + +/// Routes announced before any session exists are replayed when a session +/// reaches Established. +#[tokio::test] +async fn test_routes_replayed_on_session_up() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?; + server.local_rib.announce(route("10.100.2.0/24", LocalRouteAttrs::default())).await?; + + let mut peer = established_peer(&server).await?; + + let mut seen = vec![next_update(&mut peer).await?, next_update(&mut peer).await?] + .into_iter() + .flat_map(|u| u.announced_routes) + .map(|p| p.to_string()) + .collect::>(); + seen.sort(); + assert_eq!(seen, vec!["10.100.1.0/24", "10.100.2.0/24"]); + Ok(()) +} + +/// An announcement is fanned out to every established session. +#[tokio::test] +async fn test_announce_reaches_all_sessions() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + let mut a = established_peer(&server).await?; + let mut b = established_peer(&server).await?; + wait_for_sessions(&server, 2).await?; + + server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?; + + for peer in [&mut a, &mut b] { + let update = next_update(peer).await?; + assert_eq!(update.announced_routes, vec![prefix("10.100.1.0/24")]); + } + Ok(()) +} + +#[tokio::test] +async fn test_withdraw_sends_withdrawn_route() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + let mut peer = established_peer(&server).await?; + wait_for_sessions(&server, 1).await?; + + server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?; + let _ = next_update(&mut peer).await?; + + server.local_rib.withdraw(prefix("10.100.1.0/24"), Some("alice")).await?; + let update = next_update(&mut peer).await?; + assert_eq!(update.withdrawn_routes, vec![prefix("10.100.1.0/24")]); + assert!(update.announced_routes.is_empty()); + assert!(update.path_attributes.is_empty()); + Ok(()) +} + +/// Re-announcing with new attributes sends a fresh UPDATE (implicit replace). +#[tokio::test] +async fn test_reannounce_with_prepend_and_med() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + let mut peer = established_peer(&server).await?; + wait_for_sessions(&server, 1).await?; + + server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?; + let first = next_update(&mut peer).await?; + assert_eq!(first.path_attributes[1], AsPathAttribute::from_asns(vec![LOCAL_ASN])); + + server + .local_rib + .announce(route( + "10.100.1.0/24", + LocalRouteAttrs { prepend: 3, med: Some(42), ..LocalRouteAttrs::default() }, + )) + .await?; + let second = next_update(&mut peer).await?; + assert_eq!(second.announced_routes, vec![prefix("10.100.1.0/24")]); + assert_eq!(second.path_attributes[1], AsPathAttribute::from_asns(vec![LOCAL_ASN; 4])); + assert!(second + .path_attributes + .iter() + .any(|a| matches!(a, PathAttribute::MultiExitDisc(m) if m.0 == 42))); + Ok(()) +} + +/// The session stays healthy (keepalives keep flowing) while UPDATEs are sent. +#[tokio::test] +async fn test_keepalives_continue_alongside_updates() -> Result<()> { + let server = TestServer::start_with_hold_time(LOCAL_ASN, 3).await?; + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 1), 3).await?; + wait_for_sessions(&server, 1).await?; + + server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?; + + let mut saw_update = false; + let mut saw_keepalive = false; + tokio::time::timeout(TIMEOUT, async { + while !(saw_update && saw_keepalive) { + match peer.recv().await? { + bgp_packet::message::Message::Update(_) => saw_update = true, + bgp_packet::message::Message::KeepAlive => saw_keepalive = true, + other => eyre::bail!("unexpected {other:?}"), + } + } + Ok::<_, eyre::Error>(()) + }) + .await??; + Ok(()) +} + +// --------------------------------------------------------------------------- +// Policy +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn test_policy_rejects_prefix_outside_allowlist() -> Result<()> { + let server = TestServer::start_with_policy( + LOCAL_ASN, + LocalRibPolicy { allowed_prefixes: vec![prefix("10.100.0.0/16")], max_routes_per_owner: 2 }, + ) + .await?; + let mut peer = established_peer(&server).await?; + wait_for_sessions(&server, 1).await?; + + let err = server + .local_rib + .announce(route("192.0.2.0/24", LocalRouteAttrs::default())) + .await + .unwrap_err(); + assert_eq!(err, LocalRibError::NotAllowed(prefix("192.0.2.0/24"))); + + // An allowed prefix still goes through on the same session. + server.local_rib.announce(route("10.100.9.0/24", LocalRouteAttrs::default())).await?; + let update = next_update(&mut peer).await?; + assert_eq!(update.announced_routes, vec![prefix("10.100.9.0/24")]); + Ok(()) +} diff --git a/crates/packet/src/message.rs b/crates/packet/src/message.rs index 58daa8d..c70222f 100644 --- a/crates/packet/src/message.rs +++ b/crates/packet/src/message.rs @@ -313,7 +313,7 @@ pub enum PathAttribute { LocalPref(LocalPrefPathAttribute) = 5, AtomicAggregate(AtomicAggregatePathAttribute) = 6, Aggregator(AggregatorPathAttribute) = 7, - Communitites(CommunitiesPathAttribute) = 8, + Communities(CommunitiesPathAttribute) = 8, MpReachNlri(MpReachNlriPathAttribute) = 14, MpUnreachNlri(MpUnreachNlriPathAttribute) = 15, ExtendedCommunities(ExtendedCommunitiesPathAttribute) = 16, @@ -335,7 +335,7 @@ impl From for PathAttributeDiscriminants { v if v == Self::LocalPref as u8 => Self::LocalPref, v if v == Self::AtomicAggregate as u8 => Self::AtomicAggregate, v if v == Self::Aggregator as u8 => Self::Aggregator, - v if v == Self::Communitites as u8 => Self::Communitites, + v if v == Self::Communities as u8 => Self::Communities, v if v == Self::MpReachNlri as u8 => Self::MpReachNlri, v if v == Self::MpUnreachNlri as u8 => Self::MpUnreachNlri, v if v == Self::ExtendedCommunities as u8 => Self::ExtendedCommunities, @@ -355,7 +355,7 @@ impl PathAttribute { PathAttribute::LocalPref(_) => PathAttributeDiscriminants::LocalPref as u8, PathAttribute::AtomicAggregate(_) => PathAttributeDiscriminants::AtomicAggregate as u8, PathAttribute::Aggregator(_) => PathAttributeDiscriminants::Aggregator as u8, - PathAttribute::Communitites(_) => PathAttributeDiscriminants::Communitites as u8, + PathAttribute::Communities(_) => PathAttributeDiscriminants::Communities as u8, PathAttribute::MpReachNlri(_) => PathAttributeDiscriminants::MpReachNlri as u8, PathAttribute::MpUnreachNlri(_) => PathAttributeDiscriminants::MpUnreachNlri as u8, PathAttribute::ExtendedCommunities(_) => { @@ -377,7 +377,7 @@ impl PathAttribute { PathAttribute::LocalPref(_) => false, PathAttribute::AtomicAggregate(_) => true, PathAttribute::Aggregator(_) => true, - PathAttribute::Communitites(_) => true, + PathAttribute::Communities(_) => true, PathAttribute::MpReachNlri(_) => true, PathAttribute::MpUnreachNlri(_) => true, PathAttribute::ExtendedCommunities(_) => true, @@ -395,7 +395,7 @@ impl PathAttribute { PathAttribute::LocalPref(_) => true, PathAttribute::AtomicAggregate(_) => true, PathAttribute::Aggregator(_) => false, - PathAttribute::Communitites(_) => true, + PathAttribute::Communities(_) => true, PathAttribute::MpReachNlri(_) => false, PathAttribute::MpUnreachNlri(_) => false, PathAttribute::ExtendedCommunities(_) => true, @@ -467,8 +467,8 @@ impl PathAttribute { PathAttributeDiscriminants::Aggregator => { PathAttribute::Aggregator(AggregatorPathAttribute::from_wire(ctx, pa_buf)?.1) } - PathAttributeDiscriminants::Communitites => { - PathAttribute::Communitites(CommunitiesPathAttribute::from_wire(ctx, pa_buf)?.1) + PathAttributeDiscriminants::Communities => { + PathAttribute::Communities(CommunitiesPathAttribute::from_wire(ctx, pa_buf)?.1) } PathAttributeDiscriminants::MpReachNlri => { PathAttribute::MpReachNlri(MpReachNlriPathAttribute::from_wire(ctx, pa_buf)?.1) @@ -538,7 +538,7 @@ impl PathAttribute { PathAttribute::Aggregator(aggregator_path_attribute) => { write_path_attribute!(out, aggregator_path_attribute); } - PathAttribute::Communitites(communities_path_attribute) => { + PathAttribute::Communities(communities_path_attribute) => { write_path_attribute!(out, communities_path_attribute); } PathAttribute::MpReachNlri(mp_reach_nlri_path_attribute) => { @@ -583,7 +583,7 @@ impl PathAttribute { PathAttribute::LocalPref(a) => a.wire_len(ctx)?, PathAttribute::AtomicAggregate(a) => a.wire_len(ctx)?, PathAttribute::Aggregator(a) => a.wire_len(ctx)?, - PathAttribute::Communitites(a) => a.wire_len(ctx)?, + PathAttribute::Communities(a) => a.wire_len(ctx)?, PathAttribute::MpReachNlri(a) => a.wire_len(ctx)?, PathAttribute::MpUnreachNlri(a) => a.wire_len(ctx)?, PathAttribute::ExtendedCommunities(a) => a.wire_len(ctx)?, @@ -903,7 +903,7 @@ impl AggregatorPathAttribute { } #[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)] -pub struct CommunitiesPathAttribute(Vec<(u16, u16)>); +pub struct CommunitiesPathAttribute(pub Vec<(u16, u16)>); impl CommunitiesPathAttribute { pub fn from_wire<'a>( @@ -1624,7 +1624,7 @@ mod tests { test_path_attribute_roundtrip!( test_communities, &[0xc0, 0x08, 0x04, 0xca, 0xfe, 0xba, 0xbe], - PathAttribute::Communitites(CommunitiesPathAttribute(vec![(0xcafe, 0xbabe)])) + PathAttribute::Communities(CommunitiesPathAttribute(vec![(0xcafe, 0xbabe)])) ); // MP Reach NLRI Path Attribute. diff --git a/crates/primitives/src/ip_prefix.rs b/crates/primitives/src/ip_prefix.rs index 8e8dc09..cc29cac 100644 --- a/crates/primitives/src/ip_prefix.rs +++ b/crates/primitives/src/ip_prefix.rs @@ -34,6 +34,25 @@ impl IpPrefix { }) } + /// Returns true if `other` is equal to or more specific than `self`, i.e. + /// `self` covers `other`. Prefixes of different address families never + /// contain each other. + pub fn contains(&self, other: &IpPrefix) -> bool { + if self.address_family != other.address_family || other.length < self.length { + return false; + } + let full_bytes = (self.length / 8) as usize; + if self.prefix[..full_bytes] != other.prefix[..full_bytes] { + return false; + } + let rem_bits = self.length % 8; + if rem_bits == 0 { + return true; + } + let mask = u8::MAX << (8 - rem_bits); + (self.prefix[full_bytes] & mask) == (other.prefix[full_bytes] & mask) + } + pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<(), ToWireError> { // Verify that there is enough space to write the IpPrefix. if out.remaining() < (self.prefix.len() + 1) { @@ -205,6 +224,22 @@ mod tests { 32 ); + #[test] + fn contains_covers_more_specifics() { + let p = |s: &str| IpPrefix::try_from(s).unwrap(); + assert!(p("10.100.0.0/16").contains(&p("10.100.7.0/24"))); + assert!(p("10.100.0.0/16").contains(&p("10.100.0.0/16"))); + assert!(!p("10.100.0.0/16").contains(&p("10.101.0.0/24"))); + assert!(!p("10.100.7.0/24").contains(&p("10.100.0.0/16"))); + assert!(p("0.0.0.0/0").contains(&p("192.0.2.0/24"))); + // Non-byte-aligned boundary. + assert!(p("10.100.0.0/19").contains(&p("10.100.31.0/24"))); + assert!(!p("10.100.0.0/19").contains(&p("10.100.32.0/24"))); + // Cross-family never matches. + assert!(!p("::/0").contains(&p("10.0.0.0/8"))); + assert!(p("2001:db8::/32").contains(&p("2001:db8:cafe::/48"))); + } + verify_roundtrip!( verify_roundtrip_ipv6_0, "::/0", diff --git a/crates/rib/Cargo.toml b/crates/rib/Cargo.toml index 1b4a644..42c4a32 100644 --- a/crates/rib/Cargo.toml +++ b/crates/rib/Cargo.toml @@ -5,5 +5,7 @@ edition.workspace = true [dependencies] bgp-packet.workspace = true bgp-primitives.workspace = true +serde.workspace = true +thiserror.workspace = true tokio = { workspace = true, features = ["full"] } tracing.workspace = true diff --git a/crates/rib/src/lib.rs b/crates/rib/src/lib.rs index 8bb39e6..20fb62d 100644 --- a/crates/rib/src/lib.rs +++ b/crates/rib/src/lib.rs @@ -1,7 +1,9 @@ mod adj_rib_in; mod entry; mod event; +mod local_rib; pub use adj_rib_in::AdjRibIn; pub use entry::{PeerId, RibEntry, SnapshotEntry}; pub use event::RibEvent; +pub use local_rib::{LocalRib, LocalRibError, LocalRibEvent, LocalRibPolicy, LocalRoute, LocalRouteAttrs}; diff --git a/crates/rib/src/local_rib.rs b/crates/rib/src/local_rib.rs new file mode 100644 index 0000000..6575efb --- /dev/null +++ b/crates/rib/src/local_rib.rs @@ -0,0 +1,305 @@ +//! Locally originated routes (the "Loc-RIB" of routes this router announces +//! on its own behalf, as opposed to routes learned from peers). +//! +//! The actor stores one route per prefix together with the attributes chosen +//! by whoever announced it. It does not know about peers or sessions: the +//! server subscribes to [`LocalRibEvent`]s and turns them into UPDATEs. + +use std::collections::HashMap; +use std::time::SystemTime; + +use bgp_packet::message::{LargeCommunity, OriginPathAttribute}; +use bgp_primitives::ip_prefix::IpPrefix; +use serde::{Deserialize, Serialize}; +use tokio::sync::{broadcast, mpsc, oneshot}; +use tracing::warn; + +const EVENT_CHANNEL_CAPACITY: usize = 1024; +const CMD_CHANNEL_CAPACITY: usize = 256; + +/// User-selectable attributes for a locally originated route. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct LocalRouteAttrs { + pub origin: OriginPathAttribute, + /// Number of *additional* copies of the local ASN to prepend to the + /// AS_PATH. `0` yields the plain `[local_asn]` path. + pub prepend: u8, + pub med: Option, + /// RFC 1997 standard communities as `(high, low)` 16-bit halves. + pub communities: Vec<(u16, u16)>, + /// RFC 8092 large communities. + pub large_communities: Vec, +} + +impl Default for LocalRouteAttrs { + fn default() -> Self { + Self { + origin: OriginPathAttribute::IGP, + prepend: 0, + med: None, + communities: vec![], + large_communities: vec![], + } + } +} + +/// A route originated by this router. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct LocalRoute { + pub prefix: IpPrefix, + /// Free-form identity of whoever announced the route (e.g. a student + /// name). Used to stop one owner replacing or withdrawing another's route. + pub owner: String, + pub attrs: LocalRouteAttrs, + pub announced_at: SystemTime, +} + +impl LocalRoute { + pub fn new(prefix: IpPrefix, owner: impl Into, attrs: LocalRouteAttrs) -> Self { + Self { prefix, owner: owner.into(), attrs, announced_at: SystemTime::now() } + } +} + +/// Limits on what may be announced. Enforced inside the actor so every caller +/// (HTTP API, tests, future CLI) goes through the same checks. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct LocalRibPolicy { + /// A route is accepted only if it is covered by at least one of these + /// prefixes. Empty means no restriction. + pub allowed_prefixes: Vec, + /// Maximum number of routes a single owner may hold. `0` means unlimited. + pub max_routes_per_owner: usize, +} + +impl LocalRibPolicy { + fn allows(&self, prefix: &IpPrefix) -> bool { + self.allowed_prefixes.is_empty() || self.allowed_prefixes.iter().any(|p| p.contains(prefix)) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum LocalRibError { + #[error("prefix {0} is not covered by the allowed prefixes")] + NotAllowed(IpPrefix), + #[error("owner {owner:?} already has {limit} routes")] + OwnerLimit { owner: String, limit: usize }, + #[error("prefix {prefix} is owned by {owner:?}")] + OwnedByOther { prefix: IpPrefix, owner: String }, + #[error("local RIB actor has stopped")] + Closed, +} + +#[derive(Debug, Clone)] +pub enum LocalRibEvent { + /// A route was added or its attributes replaced. + Announced(LocalRoute), + Withdrawn { prefix: IpPrefix }, +} + +enum Command { + Announce { route: LocalRoute, reply: oneshot::Sender> }, + Withdraw { + prefix: IpPrefix, + requester: Option, + reply: oneshot::Sender, LocalRibError>>, + }, + Snapshot { reply: oneshot::Sender> }, +} + +/// Handle to the local RIB actor. Cheap to clone; all clones share one table. +#[derive(Clone)] +pub struct LocalRib { + tx: mpsc::Sender, + events: broadcast::Sender, +} + +impl LocalRib { + pub fn spawn(policy: LocalRibPolicy) -> Self { + let (tx, rx) = mpsc::channel(CMD_CHANNEL_CAPACITY); + let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY); + tokio::spawn(run_actor(rx, events.clone(), policy)); + Self { tx, events } + } + + /// Adds a route, or replaces the attributes of an existing route with the + /// same prefix and owner. + pub async fn announce(&self, route: LocalRoute) -> Result<(), LocalRibError> { + let (reply, rx) = oneshot::channel(); + self.tx + .send(Command::Announce { route, reply }) + .await + .map_err(|_| LocalRibError::Closed)?; + rx.await.map_err(|_| LocalRibError::Closed)? + } + + /// Removes a route. `requester` must match the route's owner; pass `None` + /// to withdraw regardless of owner (administrative override). Returns the + /// removed route, or `None` if the prefix was not present. + pub async fn withdraw( + &self, + prefix: IpPrefix, + requester: Option<&str>, + ) -> Result, LocalRibError> { + let (reply, rx) = oneshot::channel(); + self.tx + .send(Command::Withdraw { prefix, requester: requester.map(str::to_owned), reply }) + .await + .map_err(|_| LocalRibError::Closed)?; + rx.await.map_err(|_| LocalRibError::Closed)? + } + + /// Point-in-time copy of every route, in no particular order. + pub async fn snapshot(&self) -> Vec { + let (reply, rx) = oneshot::channel(); + let _ = self.tx.send(Command::Snapshot { reply }).await; + rx.await.unwrap_or_default() + } + + /// Subscribe to mutations. See [`crate::AdjRibIn::subscribe`] for lag + /// semantics. + pub fn subscribe(&self) -> broadcast::Receiver { + self.events.subscribe() + } +} + +async fn run_actor( + mut rx: mpsc::Receiver, + events: broadcast::Sender, + policy: LocalRibPolicy, +) { + let mut table: HashMap = HashMap::new(); + + while let Some(cmd) = rx.recv().await { + match cmd { + Command::Announce { route, reply } => { + let result = check_announce(&table, &policy, &route).map(|()| { + table.insert(route.prefix.clone(), route.clone()); + let _ = events.send(LocalRibEvent::Announced(route)); + }); + let _ = reply.send(result); + } + Command::Withdraw { prefix, requester, reply } => { + let result = match (table.get(&prefix), requester) { + (Some(existing), Some(req)) if existing.owner != req => { + Err(LocalRibError::OwnedByOther { + prefix: prefix.clone(), + owner: existing.owner.clone(), + }) + } + _ => { + let removed = table.remove(&prefix); + if removed.is_some() { + let _ = events.send(LocalRibEvent::Withdrawn { prefix }); + } + Ok(removed) + } + }; + let _ = reply.send(result); + } + Command::Snapshot { reply } => { + if reply.send(table.values().cloned().collect()).is_err() { + warn!("Snapshot reply channel closed before send"); + } + } + } + } +} + +fn check_announce( + table: &HashMap, + policy: &LocalRibPolicy, + route: &LocalRoute, +) -> Result<(), LocalRibError> { + if !policy.allows(&route.prefix) { + return Err(LocalRibError::NotAllowed(route.prefix.clone())); + } + if let Some(existing) = table.get(&route.prefix) { + if existing.owner != route.owner { + return Err(LocalRibError::OwnedByOther { + prefix: route.prefix.clone(), + owner: existing.owner.clone(), + }); + } + // Replacing our own route never changes the per-owner count. + return Ok(()); + } + if policy.max_routes_per_owner > 0 { + let held = table.values().filter(|r| r.owner == route.owner).count(); + if held >= policy.max_routes_per_owner { + return Err(LocalRibError::OwnerLimit { + owner: route.owner.clone(), + limit: policy.max_routes_per_owner, + }); + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn prefix(s: &str) -> IpPrefix { + IpPrefix::try_from(s).unwrap() + } + + fn route(p: &str, owner: &str) -> LocalRoute { + LocalRoute::new(prefix(p), owner, LocalRouteAttrs::default()) + } + + #[tokio::test] + async fn announce_and_withdraw_emit_events() { + let rib = LocalRib::spawn(LocalRibPolicy::default()); + let mut events = rib.subscribe(); + + rib.announce(route("10.0.0.0/24", "alice")).await.unwrap(); + assert!(matches!(events.recv().await.unwrap(), LocalRibEvent::Announced(r) if r.prefix == prefix("10.0.0.0/24"))); + assert_eq!(rib.snapshot().await.len(), 1); + + let removed = rib.withdraw(prefix("10.0.0.0/24"), Some("alice")).await.unwrap(); + assert_eq!(removed.unwrap().owner, "alice"); + assert!(matches!(events.recv().await.unwrap(), LocalRibEvent::Withdrawn { prefix: p } if p == prefix("10.0.0.0/24"))); + assert!(rib.snapshot().await.is_empty()); + + // Withdrawing an absent prefix is a no-op, not an error. + assert_eq!(rib.withdraw(prefix("10.0.0.0/24"), None).await.unwrap(), None); + } + + #[tokio::test] + async fn policy_allowlist_is_enforced() { + let rib = LocalRib::spawn(LocalRibPolicy { + allowed_prefixes: vec![prefix("10.100.0.0/16")], + max_routes_per_owner: 0, + }); + rib.announce(route("10.100.3.0/24", "alice")).await.unwrap(); + let err = rib.announce(route("192.0.2.0/24", "alice")).await.unwrap_err(); + assert_eq!(err, LocalRibError::NotAllowed(prefix("192.0.2.0/24"))); + } + + #[tokio::test] + async fn owner_limit_ignores_replacements() { + let rib = LocalRib::spawn(LocalRibPolicy { allowed_prefixes: vec![], max_routes_per_owner: 1 }); + rib.announce(route("10.0.0.0/24", "alice")).await.unwrap(); + // Replacing the same prefix is fine. + rib.announce(route("10.0.0.0/24", "alice")).await.unwrap(); + let err = rib.announce(route("10.0.1.0/24", "alice")).await.unwrap_err(); + assert!(matches!(err, LocalRibError::OwnerLimit { limit: 1, .. })); + // Other owners are unaffected. + rib.announce(route("10.0.1.0/24", "bob")).await.unwrap(); + } + + #[tokio::test] + async fn other_owner_cannot_replace_or_withdraw() { + let rib = LocalRib::spawn(LocalRibPolicy::default()); + rib.announce(route("10.0.0.0/24", "alice")).await.unwrap(); + + let err = rib.announce(route("10.0.0.0/24", "bob")).await.unwrap_err(); + assert!(matches!(err, LocalRibError::OwnedByOther { .. })); + + let err = rib.withdraw(prefix("10.0.0.0/24"), Some("bob")).await.unwrap_err(); + assert!(matches!(err, LocalRibError::OwnedByOther { .. })); + + // Administrative override works. + assert!(rib.withdraw(prefix("10.0.0.0/24"), None).await.unwrap().is_some()); + } +} diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index e1a5686..9ba77ba 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -15,6 +15,7 @@ tokio.workspace = true tracing-subscriber.workspace = true tracing.workspace = true protobuf.workspace = true +thiserror.workspace = true [build-dependencies] protobuf-codegen.workspace = true diff --git a/crates/server/proto/config.proto b/crates/server/proto/config.proto index 3210367..37fc5ab 100644 --- a/crates/server/proto/config.proto +++ b/crates/server/proto/config.proto @@ -17,6 +17,20 @@ message BgpConfig { // Peers configured on this server. repeated Peer peers = 4; + + // Limits on routes originated locally (e.g. via the demo UI). + LocalRoutePolicy local_route_policy = 5; +} + +// Limits on locally originated routes. +message LocalRoutePolicy { + // A locally originated route is accepted only if it is covered by one of + // these prefixes, written in CIDR notation (e.g. "10.100.0.0/16"). + // Empty means any prefix may be announced. + repeated string allowed_prefixes = 1; + + // Maximum number of routes a single owner may announce. 0 means unlimited. + uint32 max_routes_per_owner = 2; } // An IP address and port pair. diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs new file mode 100644 index 0000000..3be879a --- /dev/null +++ b/crates/server/src/advertise.rs @@ -0,0 +1,267 @@ +//! Turns locally originated routes into UPDATE messages for every established +//! session. +//! +//! The advertiser task listens for [`LocalRibEvent`]s and [`SessionEvent`]s. +//! A route change is fanned out to all live sessions; a new session receives +//! the full local table. Sending the same route twice is harmless (BGP treats +//! a re-announcement as an implicit replace), so the two event streams need +//! no coordination beyond being handled by one task. + +use std::net::IpAddr; + +use bgp_packet::message::{ + AsPathAttribute, CommunitiesPathAttribute, LargeCommunitiesPathAttribute, Message, + MpReachNlriPathAttribute, MpUnreachNlriPathAttribute, MultiExitDiscPathAttribute, + NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage, +}; +use bgp_primitives::constants::{AddressFamilyId, SubsequentAfi}; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rib::{LocalRib, LocalRibEvent, LocalRoute}; +use tokio::sync::broadcast::error::RecvError; +use tokio::task::JoinHandle; +use tracing::{debug, info, warn}; + +use crate::session::{SessionEvent, SessionHandle, SessionRegistry}; + +pub fn spawn_advertiser( + local_rib: LocalRib, + registry: SessionRegistry, + local_asn: u32, +) -> JoinHandle<()> { + tokio::spawn(run(local_rib, registry, local_asn)) +} + +async fn run(local_rib: LocalRib, registry: SessionRegistry, local_asn: u32) { + let mut rib_events = local_rib.subscribe(); + let mut session_events = registry.subscribe(); + + loop { + tokio::select! { + ev = rib_events.recv() => match ev { + Ok(LocalRibEvent::Announced(route)) => { + for session in registry.list() { + send_route(&session, &route, local_asn).await; + } + } + Ok(LocalRibEvent::Withdrawn { prefix }) => { + for session in registry.list() { + send_withdraw(&session, &prefix).await; + } + } + Err(RecvError::Lagged(n)) => { + warn!(missed = n, "Advertiser lagged on local RIB events, resyncing all sessions"); + resync_all(&local_rib, ®istry, local_asn).await; + } + Err(RecvError::Closed) => return, + }, + ev = session_events.recv() => match ev { + Ok(SessionEvent::Up(session)) => { + let routes = local_rib.snapshot().await; + info!(peer = %session.peer, routes = routes.len(), "Session up, sending local routes"); + for route in &routes { + send_route(&session, route, local_asn).await; + } + } + Ok(SessionEvent::Down(_)) => {} + Err(RecvError::Lagged(n)) => { + warn!(missed = n, "Advertiser lagged on session events, resyncing all sessions"); + resync_all(&local_rib, ®istry, local_asn).await; + } + Err(RecvError::Closed) => return, + }, + } + } +} + +async fn resync_all(local_rib: &LocalRib, registry: &SessionRegistry, local_asn: u32) { + let routes = local_rib.snapshot().await; + for session in registry.list() { + for route in &routes { + send_route(&session, route, local_asn).await; + } + } +} + +async fn send_route(session: &SessionHandle, route: &LocalRoute, local_asn: u32) { + match build_announce(route, local_asn, session.local_addr.ip()) { + Some(update) => { + debug!(peer = %session.peer, prefix = %route.prefix, "Announcing local route"); + if let Err(e) = session.send(Message::Update(update)).await { + warn!(peer = %session.peer, err = %e, "Failed to queue announce"); + } + } + None => debug!( + peer = %session.peer, + prefix = %route.prefix, + "Skipping route: address family does not match session transport" + ), + } +} + +async fn send_withdraw(session: &SessionHandle, prefix: &IpPrefix) { + debug!(peer = %session.peer, %prefix, "Withdrawing local route"); + if let Err(e) = session.send(Message::Update(build_withdraw(prefix))).await { + warn!(peer = %session.peer, err = %e, "Failed to queue withdraw"); + } +} + +/// Builds the UPDATE announcing `route` with `next_hop` as the next hop. +/// +/// Returns `None` when the prefix family does not match the next-hop family: +/// carrying IPv4 NLRI over an IPv6 next hop (RFC 8950) or vice versa is not +/// supported yet. +pub fn build_announce(route: &LocalRoute, local_asn: u32, next_hop: IpAddr) -> Option { + let attrs = &route.attrs; + + // Path attributes in ascending type-code order, as most implementations + // expect them. + let mut path_attributes = vec![ + PathAttribute::Origin(attrs.origin), + AsPathAttribute::from_asns(vec![local_asn; 1 + attrs.prepend as usize]), + ]; + + let mut announced_routes = vec![]; + match (route.prefix.address_family, next_hop) { + (AddressFamilyId::Ipv4, IpAddr::V4(nh)) => { + path_attributes.push(PathAttribute::NextHop(NextHopPathAttribute(nh))); + announced_routes.push(route.prefix.clone()); + } + (AddressFamilyId::Ipv6, IpAddr::V6(_)) => {} + _ => return None, + } + + if let Some(med) = attrs.med { + path_attributes.push(PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(med))); + } + if !attrs.communities.is_empty() { + path_attributes.push(PathAttribute::Communities(CommunitiesPathAttribute( + attrs.communities.clone(), + ))); + } + if let (AddressFamilyId::Ipv6, IpAddr::V6(nh)) = (route.prefix.address_family, next_hop) { + path_attributes.push(PathAttribute::MpReachNlri(MpReachNlriPathAttribute { + afi: AddressFamilyId::Ipv6, + safi: SubsequentAfi::Unicast, + next_hop: NlriNextHop::Ipv6(nh), + prefixes: vec![route.prefix.clone()], + })); + } + if !attrs.large_communities.is_empty() { + path_attributes.push(PathAttribute::LargeCommunities(LargeCommunitiesPathAttribute { + communities: attrs.large_communities.clone(), + })); + } + + Some(UpdateMessage { withdrawn_routes: vec![], path_attributes, announced_routes }) +} + +/// Builds the UPDATE withdrawing `prefix`. +pub fn build_withdraw(prefix: &IpPrefix) -> UpdateMessage { + match prefix.address_family { + AddressFamilyId::Ipv4 => UpdateMessage { + withdrawn_routes: vec![prefix.clone()], + path_attributes: vec![], + announced_routes: vec![], + }, + AddressFamilyId::Ipv6 => UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute { + afi: AddressFamilyId::Ipv6, + safi: SubsequentAfi::Unicast, + prefixes: vec![prefix.clone()], + })], + 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() + } + + #[test] + fn ipv4_announce_has_expected_attributes_and_roundtrips() { + let route = LocalRoute::new( + prefix("10.100.1.0/24"), + "alice", + 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, 65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1))).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: Some(true), address_family: None }; + let msg = Message::Update(update.clone()); + let mut buf = BytesMut::new(); + msg.to_wire(&ctx, &mut buf).unwrap(); + let (_, parsed) = Message::from_wire(&ctx, &buf).unwrap(); + let Message::Update(parsed) = parsed else { panic!("expected UPDATE") }; + assert_eq!(parsed, update); + } + + #[test] + fn ipv6_announce_uses_mp_reach() { + let route = LocalRoute::new(prefix("2001:db8:1::/48"), "alice", LocalRouteAttrs::default()); + let nh = Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1); + let update = build_announce(&route, 65000, IpAddr::V6(nh)).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")] + ))); + } + + #[test] + fn family_mismatch_is_skipped() { + let v6 = LocalRoute::new(prefix("2001:db8::/32"), "a", LocalRouteAttrs::default()); + assert!(build_announce(&v6, 65000, IpAddr::V4(Ipv4Addr::LOCALHOST)).is_none()); + let v4 = LocalRoute::new(prefix("10.0.0.0/8"), "a", LocalRouteAttrs::default()); + assert!(build_announce(&v4, 65000, IpAddr::V6(Ipv6Addr::LOCALHOST)).is_none()); + } + + #[test] + fn withdraw_by_family() { + let v4 = build_withdraw(&prefix("10.0.0.0/8")); + 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")); + assert!(v6.withdrawn_routes.is_empty()); + assert!(matches!(&v6.path_attributes[..], [PathAttribute::MpUnreachNlri(m)] if m.prefixes == vec![prefix("2001:db8::/32")])); + } +} diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index 317ea7a..d15f872 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -1,7 +1,9 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::str::FromStr; -use eyre::{Result, bail, eyre}; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rib::LocalRibPolicy; +use eyre::{Result, WrapErr, bail, eyre}; use crate::proto::config::BgpConfig; @@ -16,6 +18,8 @@ pub struct Config { pub hold_time_secs: u16, pub listen_endpoints: Vec, pub peers: Vec, + /// Limits on locally originated routes. + pub local_route_policy: LocalRibPolicy, } pub struct PeerConfig { @@ -74,7 +78,26 @@ impl Config { }) .collect::>()?; - Ok(Self { router_id, local_asn: proto.local_asn, hold_time_secs: DEFAULT_HOLD_TIME_SECS, listen_endpoints, peers }) + let local_route_policy = match proto.local_route_policy.into_option() { + Some(policy) => LocalRibPolicy { + allowed_prefixes: policy + .allowed_prefixes + .iter() + .map(|p| IpPrefix::try_from(p.as_str()).wrap_err_with(|| format!("Invalid allowed prefix {p:?}"))) + .collect::>()?, + max_routes_per_owner: policy.max_routes_per_owner as usize, + }, + None => LocalRibPolicy::default(), + }; + + Ok(Self { + router_id, + local_asn: proto.local_asn, + hold_time_secs: DEFAULT_HOLD_TIME_SECS, + listen_endpoints, + peers, + local_route_policy, + }) } } diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index 7a9ba48..d9b24ab 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -12,11 +12,18 @@ use eyre::{Result, bail, eyre}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::sync::mpsc; use tokio::time::{Instant, interval_at, sleep_until}; use tracing::{info, warn}; use bgp_packet::message::{AS_TRANS, BGP4_VERSION}; +use crate::session::{SessionHandle, SessionRegistry}; + +/// Number of outbound messages that may be queued for a session before +/// senders are back-pressured. +const OUTBOUND_QUEUE_CAPACITY: usize = 1024; + /// Minimal per-session config extracted from the server config. #[derive(Clone)] pub struct SessionConfig { @@ -24,6 +31,7 @@ pub struct SessionConfig { pub local_asn: u32, pub hold_time_secs: u16, pub adj_rib_in: AdjRibIn, + pub registry: SessionRegistry, } /// Drive the BGP FSM for a single accepted TCP connection. @@ -34,6 +42,7 @@ pub async fn run_session( peer_addr: SocketAddr, config: SessionConfig, ) -> Result<()> { + let local_addr = stream.local_addr()?; let (mut reader, mut writer) = stream.into_split(); // --- OpenSent ----------------------------------------------------------- @@ -136,6 +145,16 @@ pub async fn run_session( let peer_id = PeerId { addr: peer_addr, asn: peer_asn }; // --- Established -------------------------------------------------------- + // Register so the rest of the server can see this session and queue + // messages for it. The guard deregisters on any exit path. + let (outbound_tx, mut outbound_rx) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY); + let _guard = config.registry.register(SessionHandle::new( + peer_id, + local_addr, + peer_four_byte_asn, + outbound_tx, + )); + established_loop( &mut reader, &mut writer, @@ -143,6 +162,7 @@ pub async fn run_session( negotiated_hold_time, peer_id, &config.adj_rib_in, + &mut outbound_rx, ) .await } @@ -154,14 +174,15 @@ async fn established_loop( hold_time_secs: u16, peer_id: PeerId, adj_rib_in: &AdjRibIn, + outbound: &mut mpsc::Receiver, ) -> Result<()> { - // When hold_time is 0, keepalives and the hold timer are disabled (RFC 4271 §4.4). - if hold_time_secs == 0 { - return no_timer_loop(reader, writer, ctx, peer_id, adj_rib_in).await; - } - - let hold_duration = Duration::from_secs(hold_time_secs as u64); - let keepalive_interval = Duration::from_secs(hold_time_secs as u64 / 3); + // When hold_time is 0, keepalives and the hold timer are disabled + // (RFC 4271 §4.4). The timer branches below are gated off in that case; + // the placeholder durations are never observed. + let timers_enabled = hold_time_secs != 0; + let secs = if timers_enabled { hold_time_secs as u64 } else { 3 }; + let hold_duration = Duration::from_secs(secs); + let keepalive_interval = Duration::from_secs(secs / 3); // interval_at avoids firing immediately; first keepalive is sent after one interval. let mut keepalive_timer = interval_at(Instant::now() + keepalive_interval, keepalive_interval); @@ -178,6 +199,7 @@ async fn established_loop( hold_timer.as_mut().reset(Instant::now() + hold_duration); } Message::Update(update) => { + hold_timer.as_mut().reset(Instant::now() + hold_duration); info!( peer = %peer_id, withdrawn = update.withdrawn_routes.len(), @@ -195,13 +217,26 @@ async fn established_loop( send_notification(writer, BgpError::FsmError(FsmSubcode::UnexpectedEstablished), ctx).await; bail!("Received OPEN in Established state from {}", peer_id); } - Message::RouteRefresh => {} + Message::RouteRefresh => { + hold_timer.as_mut().reset(Instant::now() + hold_duration); + } } } - _ = keepalive_timer.tick() => { + Some(msg) = outbound.recv() => { + if let Message::Update(update) = &msg { + info!( + peer = %peer_id, + withdrawn = update.withdrawn_routes.len(), + announced = update.announced_routes.len(), + "Sending UPDATE", + ); + } + send_message(writer, &msg, ctx).await?; + } + _ = keepalive_timer.tick(), if timers_enabled => { send_message(writer, &Message::KeepAlive, ctx).await?; } - _ = &mut hold_timer => { + _ = &mut hold_timer, if timers_enabled => { warn!(peer = %peer_id, "Hold timer expired"); send_notification(writer, BgpError::HoldTimer, ctx).await; bail!("Hold timer expired for peer {}", peer_id); @@ -210,45 +245,6 @@ async fn established_loop( } } -/// Hold-timer-disabled path for hold_time == 0. -async fn no_timer_loop( - reader: &mut OwnedReadHalf, - writer: &mut OwnedWriteHalf, - ctx: &ParserContext, - peer_id: PeerId, - adj_rib_in: &AdjRibIn, -) -> Result<()> { - loop { - match read_message(reader, ctx).await? { - Message::KeepAlive => {} - Message::Update(update) => { - info!( - peer = %peer_id, - withdrawn = update.withdrawn_routes.len(), - announced = update.announced_routes.len(), - "Received UPDATE", - ); - apply_update(adj_rib_in, peer_id, &update.path_attributes, - &update.withdrawn_routes, &update.announced_routes).await; - } - Message::Notification(notif) => { - info!(peer = %peer_id, error = %notif.error, "Received NOTIFICATION, closing session"); - return Ok(()); - } - Message::Open(_) => { - send_notification( - writer, - BgpError::FsmError(FsmSubcode::UnexpectedEstablished), - ctx, - ) - .await; - bail!("Received OPEN in Established state from {}", peer_id); - } - Message::RouteRefresh => {} - } - } -} - // --------------------------------------------------------------------------- // Wire helpers // --------------------------------------------------------------------------- diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index e88955b..ec103bf 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -1,7 +1,9 @@ +pub mod advertise; pub mod config; pub mod connection; pub mod fsm; pub mod server; +pub mod session; pub mod tracing; pub mod proto { diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index 69f791c..132922c 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -7,11 +7,13 @@ use futures::stream::{FuturesUnordered, StreamExt}; use tokio::net::{TcpListener, TcpStream}; use tracing::{error, info}; -use bgp_rib::AdjRibIn; +use bgp_rib::{AdjRibIn, LocalRib}; +use crate::advertise::spawn_advertiser; use crate::config::Config; use crate::connection::handle_connection; use crate::fsm::SessionConfig; +use crate::session::SessionRegistry; pub struct BgpServerBuilder { config: Config, @@ -25,6 +27,10 @@ pub struct BgpServer { /// Shared RIB handle — created at bind time so callers can clone it before /// the accept loop starts (e.g. tests that want to introspect the RIB). pub adj_rib_in: AdjRibIn, + /// Routes this router originates; announced to every established peer. + pub local_rib: LocalRib, + /// Live view of established sessions. + pub registry: SessionRegistry, } impl BgpServerBuilder { @@ -46,9 +52,11 @@ impl BgpServerBuilder { listeners.push((bound_addr, listener)); } Ok(BgpServer { - config: self.config, listeners, adj_rib_in: AdjRibIn::spawn(), + local_rib: LocalRib::spawn(self.config.local_route_policy.clone()), + registry: SessionRegistry::new(), + config: self.config, }) } @@ -77,8 +85,15 @@ impl BgpServer { local_asn: self.config.local_asn, hold_time_secs: self.config.hold_time_secs, adj_rib_in: self.adj_rib_in.clone(), + registry: self.registry.clone(), }); + let _advertiser = spawn_advertiser( + self.local_rib.clone(), + self.registry.clone(), + self.config.local_asn, + ); + let mut accepting: FuturesUnordered<_> = self .listeners .into_iter() diff --git a/crates/server/src/session.rs b/crates/server/src/session.rs new file mode 100644 index 0000000..8a84f9a --- /dev/null +++ b/crates/server/src/session.rs @@ -0,0 +1,118 @@ +//! Registry of established BGP sessions. +//! +//! A session registers itself when it reaches Established and is removed when +//! its [`SessionGuard`] drops, so a session that errors out is always +//! deregistered. Other parts of the server (the advertiser, the HTTP API) use +//! the registry to enumerate live sessions and queue outbound messages. + +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::{Arc, RwLock}; +use std::time::SystemTime; + +use bgp_packet::message::Message; +use bgp_rib::PeerId; +use tokio::sync::{broadcast, mpsc}; + +const EVENT_CHANNEL_CAPACITY: usize = 256; + +/// Handle to one established session. Cloning is cheap. +#[derive(Debug, Clone)] +pub struct SessionHandle { + pub peer: PeerId, + /// Our side of the TCP connection; used as the next hop for routes we + /// originate to this peer. + pub local_addr: SocketAddr, + pub established_at: SystemTime, + /// Whether the peer negotiated RFC 6793 four-octet ASNs. Determines how + /// AS_PATH is encoded on the wire for this session. + pub four_octet_asn: bool, + outbound: mpsc::Sender, +} + +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[error("session with {0} has closed")] +pub struct SessionClosed(pub PeerId); + +impl SessionHandle { + pub fn new( + peer: PeerId, + local_addr: SocketAddr, + four_octet_asn: bool, + outbound: mpsc::Sender, + ) -> Self { + Self { peer, local_addr, established_at: SystemTime::now(), four_octet_asn, outbound } + } + + /// Queues a message for transmission on this session. Waits for queue + /// space if the session's writer is behind. + pub async fn send(&self, msg: Message) -> Result<(), SessionClosed> { + self.outbound.send(msg).await.map_err(|_| SessionClosed(self.peer)) + } +} + +#[derive(Debug, Clone)] +pub enum SessionEvent { + Up(SessionHandle), + Down(PeerId), +} + +#[derive(Clone)] +pub struct SessionRegistry { + inner: Arc>>, + events: broadcast::Sender, +} + +impl Default for SessionRegistry { + fn default() -> Self { + Self::new() + } +} + +impl SessionRegistry { + pub fn new() -> Self { + let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY); + Self { inner: Arc::new(RwLock::new(HashMap::new())), events } + } + + /// Adds a session and emits [`SessionEvent::Up`]. The returned guard + /// removes the session again when dropped. + pub fn register(&self, handle: SessionHandle) -> SessionGuard { + let peer = handle.peer; + self.inner.write().expect("session registry poisoned").insert(peer, handle.clone()); + let _ = self.events.send(SessionEvent::Up(handle)); + SessionGuard { registry: self.clone(), peer } + } + + pub fn get(&self, peer: &PeerId) -> Option { + self.inner.read().expect("session registry poisoned").get(peer).cloned() + } + + /// All currently established sessions, in no particular order. + pub fn list(&self) -> Vec { + self.inner.read().expect("session registry poisoned").values().cloned().collect() + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.events.subscribe() + } + + fn remove(&self, peer: &PeerId) { + let removed = self.inner.write().expect("session registry poisoned").remove(peer); + if removed.is_some() { + let _ = self.events.send(SessionEvent::Down(*peer)); + } + } +} + +/// Removes the session from its registry on drop. +pub struct SessionGuard { + registry: SessionRegistry, + peer: PeerId, +} + +impl Drop for SessionGuard { + fn drop(&mut self) { + self.registry.remove(&self.peer); + } +} diff --git a/docs/RFCs.md b/docs/RFCs.md index b37f359..3f69462 100644 --- a/docs/RFCs.md +++ b/docs/RFCs.md @@ -27,7 +27,10 @@ The base protocol specification. | §8 | FSM — passive (inbound) side: OpenSent → OpenConfirm → Established | ✅ | | §8 | FSM — active (outbound) side: Idle → Connect → Active | ❌ | | §9 | Decision process / best-path selection | ❌ | -| §9 | RIB (Adj-RIB-In, Loc-RIB, Adj-RIB-Out) | ❌ | +| §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) | ❌ | +| §9.2 | Hold timer reset on any received message (not only KEEPALIVE) | ✅ | --- @@ -122,6 +125,7 @@ https://datatracker.ietf.org/doc/html/rfc1997 | Feature | Status | |---------|--------| | `COMMUNITIES` path attribute (parse, serialise) | ✅ | +| Attach to locally originated routes | ✅ | ### RFC 4360 — BGP Extended Communities Attribute https://datatracker.ietf.org/doc/html/rfc4360 @@ -137,6 +141,7 @@ https://datatracker.ietf.org/doc/html/rfc8092 | Feature | Status | |---------|--------| | `LARGE_COMMUNITIES` path attribute (parse, serialise) | ✅ | +| Attach to locally originated routes | ✅ | --- diff --git a/testdata/server.textpb b/testdata/server.textpb index 1985ed9..c9bfb62 100644 --- a/testdata/server.textpb +++ b/testdata/server.textpb @@ -8,3 +8,8 @@ listen_endpoints { address: "[::1]" port: 179 } +local_route_policy { + allowed_prefixes: "10.100.0.0/16" + allowed_prefixes: "2001:db8:100::/48" + max_routes_per_owner: 4 +}