diff --git a/Cargo.lock b/Cargo.lock index b3f3785..bb01e18 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -111,6 +111,19 @@ dependencies = [ "tokio", ] +[[package]] +name = "bgp-integration-tests" +version = "0.0.0" +dependencies = [ + "bgp-packet", + "bgp-primitives", + "bgp-rib", + "bgp-server", + "bytes", + "eyre", + "tokio", +] + [[package]] name = "bgp-packet" version = "0.0.2" @@ -140,12 +153,27 @@ dependencies = [ "thiserror 2.0.12", ] +[[package]] +name = "bgp-rib" +version = "0.0.0" +dependencies = [ + "bgp-packet", + "bgp-primitives", + "tokio", + "tracing", +] + [[package]] name = "bgp-server" version = "0.0.0" dependencies = [ + "bgp-packet", + "bgp-primitives", + "bgp-rib", + "bytes", "clap", "eyre", + "futures", "prometheus", "protobuf", "protobuf-codegen", @@ -282,6 +310,94 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "futures" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-channel" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" +dependencies = [ + "futures-core", + "futures-sink", +] + +[[package]] +name = "futures-core" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" + +[[package]] +name = "futures-executor" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" + +[[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "futures-sink" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893" + +[[package]] +name = "futures-task" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" + +[[package]] +name = "futures-util" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" +dependencies = [ + "futures-channel", + "futures-core", + "futures-io", + "futures-macro", + "futures-sink", + "futures-task", + "memchr", + "pin-project-lite", + "slab", +] + [[package]] name = "getrandom" version = "0.3.4" @@ -630,7 +746,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.4.15", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -707,6 +823,12 @@ dependencies = [ "libc", ] +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + [[package]] name = "smallvec" version = "1.14.0" diff --git a/Cargo.toml b/Cargo.toml index 290451b..cefb67f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,7 +17,9 @@ version = "0.0.2" # --- Our crates --- bgp-packet = { path = "crates/packet" } bgp-primitives = { path = "crates/primitives" } +bgp-rib = { path = "crates/rib" } bgp-server = { path = "crates/server" } +bgp-integration-tests = { path = "crates/integration-tests" } # --- General --- bitfield = "0.19.0" @@ -36,3 +38,4 @@ prometheus = "0.14.0" protobuf = { package = "protobuf", version = "3.7.2" } protobuf-codegen = { package = "protobuf-codegen", version = "3.7.2" } clap = { version = "4.5.53", features = ["derive"] } +futures = "0.3" diff --git a/crates/integration-tests/Cargo.toml b/crates/integration-tests/Cargo.toml new file mode 100644 index 0000000..8a08c69 --- /dev/null +++ b/crates/integration-tests/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "bgp-integration-tests" +edition.workspace = true +publish = false + +[dependencies] +bgp-packet.workspace = true +bgp-primitives.workspace = true +bgp-rib.workspace = true +bgp-server.workspace = true +bytes.workspace = true +eyre.workspace = true +tokio.workspace = true diff --git a/crates/integration-tests/src/lib.rs b/crates/integration-tests/src/lib.rs new file mode 100644 index 0000000..cedbe12 --- /dev/null +++ b/crates/integration-tests/src/lib.rs @@ -0,0 +1,2 @@ +pub mod test_peer; +pub mod test_server; diff --git a/crates/integration-tests/src/test_peer.rs b/crates/integration-tests/src/test_peer.rs new file mode 100644 index 0000000..fc54ba7 --- /dev/null +++ b/crates/integration-tests/src/test_peer.rs @@ -0,0 +1,127 @@ +use std::net::{Ipv4Addr, SocketAddr}; + +use bgp_packet::message::{BGP_HEADER_LEN, Message}; +use bgp_packet::open::{Capability, OpenMessage, OpenOption}; +use bgp_primitives::constants::{AddressFamilyId, SubsequentAfi}; +use bgp_primitives::parser::ParserContext; +use bytes::BytesMut; +use eyre::{Result, bail, eyre}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; + +/// A minimal BGP peer for use in tests. Connects over TCP and can exchange +/// framed BGP messages using the `bgp-packet` wire format. +pub struct TestPeer { + stream: TcpStream, + pub ctx: ParserContext, +} + +impl TestPeer { + pub async fn connect(addr: SocketAddr) -> Result { + let stream = TcpStream::connect(addr).await?; + Ok(Self { stream, ctx: ParserContext::default() }) + } + + /// Send a BGP message. + pub async fn send(&mut self, msg: &Message) -> Result<()> { + let mut buf = BytesMut::new(); + msg.to_wire(&self.ctx, &mut buf)?; + self.stream.write_all(&buf).await?; + Ok(()) + } + + /// Read exactly one BGP message. + pub async fn recv(&mut self) -> Result { + let mut header = [0u8; BGP_HEADER_LEN as usize]; + self.stream.read_exact(&mut header).await?; + + let total_len = u16::from_be_bytes([header[16], header[17]]) as usize; + if total_len < BGP_HEADER_LEN as usize { + bail!("Received BGP message with invalid length {total_len}"); + } + + let mut buf = vec![0u8; total_len]; + buf[..BGP_HEADER_LEN as usize].copy_from_slice(&header); + self.stream.read_exact(&mut buf[BGP_HEADER_LEN as usize..]).await?; + + let (_, msg) = Message::from_wire(&self.ctx, &buf) + .map_err(|e| eyre!("Failed to parse BGP message: {e:?}"))?; + Ok(msg) + } + + /// Receive and unwrap as `OpenMessage`, failing if a different type arrives. + pub async fn recv_open(&mut self) -> Result { + match self.recv().await? { + Message::Open(o) => { + // Enable 4-octet ASN parsing if the peer advertised it. + let four_byte = o.options.iter().any(|opt| { + matches!(opt, OpenOption::Capabilities(caps) + if caps.iter().any(|c| matches!(c, Capability::FourByteAsn { .. }))) + }); + self.ctx.four_octet_asn = Some(four_byte); + Ok(o) + } + other => bail!("Expected OPEN, got {:?}", other), + } + } + + /// Send an OPEN message with the given parameters and standard capabilities. + pub async fn send_open( + &mut self, + local_asn: u32, + router_id: Ipv4Addr, + hold_time: u16, + ) -> Result<()> { + let asn_field = if local_asn > u16::MAX as u32 { 23456 } else { local_asn as u16 }; + let open = Message::Open(OpenMessage { + version: 4, + asn: asn_field, + hold_time, + identifier: router_id, + options: vec![OpenOption::Capabilities(vec![ + Capability::FourByteAsn { asn: local_asn }, + Capability::MultiProtocol { + afi: AddressFamilyId::Ipv4, + safi: SubsequentAfi::Unicast, + }, + Capability::RouteRefresh, + ])], + }); + self.send(&open).await + } + + /// Complete a full OPEN/KEEPALIVE handshake as the peer side. + /// Returns the server's OPEN message. + pub async fn handshake( + &mut self, + local_asn: u32, + router_id: Ipv4Addr, + hold_time: u16, + ) -> Result { + self.send_open(local_asn, router_id, hold_time).await?; + let server_open = self.recv_open().await?; + // Expect server KEEPALIVE + match self.recv().await? { + Message::KeepAlive => {} + other => bail!("Expected KEEPALIVE from server, got {:?}", other), + } + // Send our KEEPALIVE + self.send(&Message::KeepAlive).await?; + Ok(server_open) + } + + /// Send raw bytes without framing — useful for testing invalid input. + pub async fn send_raw(&mut self, bytes: &[u8]) -> Result<()> { + self.stream.write_all(bytes).await?; + Ok(()) + } + + /// Returns true if the remote end has closed the connection. + pub async fn is_closed(&mut self) -> bool { + let mut buf = [0u8; 1]; + match self.stream.read(&mut buf).await { + Ok(0) | Err(_) => true, + Ok(_) => false, + } + } +} diff --git a/crates/integration-tests/src/test_server.rs b/crates/integration-tests/src/test_server.rs new file mode 100644 index 0000000..d41fa4d --- /dev/null +++ b/crates/integration-tests/src/test_server.rs @@ -0,0 +1,56 @@ +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; + +use bgp_rib::AdjRibIn; +use bgp_server::config::Config; +use bgp_server::server::BgpServerBuilder; +use eyre::Result; +use tokio::task::JoinHandle; + +/// A BGP server started on an ephemeral port for testing. +/// Aborts the server task when dropped. +pub struct TestServer { + handle: JoinHandle>, + /// The single IPv4 loopback address the server is bound to. + pub addr: SocketAddr, + /// Handle to the server's Adj-RIB-In for test introspection. + pub adj_rib_in: AdjRibIn, +} + +impl TestServer { + /// Start a server with `local_asn` and the default 90 s hold time. + pub async fn start(local_asn: u32) -> Result { + Self::start_with_hold_time(local_asn, 90).await + } + + /// 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 + } + + async fn start_inner(local_asn: u32, hold_time_secs: u16) -> 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![], + }; + + let bound = BgpServerBuilder::new(config).bind().await?; + let addr = bound + .local_addrs() + .into_iter() + .next() + .expect("at least one listener"); + let adj_rib_in = bound.adj_rib_in.clone(); + + let handle = tokio::spawn(bound.run()); + Ok(Self { handle, addr, adj_rib_in }) + } +} + +impl Drop for TestServer { + fn drop(&mut self) { + self.handle.abort(); + } +} diff --git a/crates/integration-tests/tests/basic_connection.rs b/crates/integration-tests/tests/basic_connection.rs new file mode 100644 index 0000000..c69ed82 --- /dev/null +++ b/crates/integration-tests/tests/basic_connection.rs @@ -0,0 +1,376 @@ +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::errors::{BgpError, CeaseSubcode}; +use bgp_packet::message::{ + Message, MpReachNlriPathAttribute, MpUnreachNlriPathAttribute, NlriNextHop, + NotificationMessage, PathAttribute, UpdateMessage, +}; +use bgp_primitives::constants::{AddressFamilyId, SubsequentAfi}; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rib::RibEvent; +use eyre::Result; + +// --------------------------------------------------------------------------- +// Connectivity +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn test_server_starts() -> Result<()> { + let server = TestServer::start(65000).await?; + assert_ne!(server.addr.port(), 0); + Ok(()) +} + +#[tokio::test] +async fn test_server_accepts_tcp_connection() -> Result<()> { + let server = TestServer::start(65000).await?; + let _peer = TestPeer::connect(server.addr).await?; + Ok(()) +} + +#[tokio::test] +async fn test_server_accepts_multiple_connections() -> Result<()> { + let server = TestServer::start(65000).await?; + let mut peers = Vec::new(); + for _ in 0..5 { + peers.push(TestPeer::connect(server.addr).await?); + } + assert_eq!(peers.len(), 5); + Ok(()) +} + +// --------------------------------------------------------------------------- +// OpenSent state +// --------------------------------------------------------------------------- + +/// Server sends an OPEN immediately after TCP is established. +#[tokio::test] +async fn test_server_sends_open() -> Result<()> { + let server = TestServer::start(65000).await?; + let mut peer = TestPeer::connect(server.addr).await?; + + let open = peer.recv_open().await?; + assert_eq!(open.version, 4); + assert_eq!(open.asn, 65000); + Ok(()) +} + +/// Server sends a NOTIFICATION and closes if we send garbage instead of OPEN. +#[tokio::test] +async fn test_bad_marker_does_not_crash_server() -> Result<()> { + let server = TestServer::start(65000).await?; + + let mut bad = TestPeer::connect(server.addr).await?; + // Discard the server's OPEN, then send garbage. + let _ = bad.recv_open().await?; + bad.send_raw(&[0x00; 19]).await?; + drop(bad); + + // New connections are still accepted. + let _good = TestPeer::connect(server.addr).await?; + Ok(()) +} + +// --------------------------------------------------------------------------- +// OpenConfirm → Established +// --------------------------------------------------------------------------- + +/// Full OPEN/KEEPALIVE handshake reaches Established. +#[tokio::test] +async fn test_session_established() -> Result<()> { + let server = TestServer::start(65000).await?; + let mut peer = TestPeer::connect(server.addr).await?; + + let server_open = peer + .handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90) + .await?; + + assert_eq!(server_open.version, 4); + Ok(()) +} + +/// After Established, the server sends periodic KEEPALIVEs. +#[tokio::test] +async fn test_server_sends_keepalives() -> Result<()> { + // Use a short hold time so the keepalive interval (hold/3 = 1 s) fires quickly. + let server = TestServer::start_with_hold_time(65000, 3).await?; + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 3).await?; + + // We should receive a KEEPALIVE within ~1 second (hold/3). + let msg = tokio::time::timeout(Duration::from_secs(3), peer.recv()).await??; + assert!( + matches!(msg, Message::KeepAlive), + "Expected KEEPALIVE, got {:?}", + msg + ); + Ok(()) +} + +/// Server closes with a NOTIFICATION when the hold timer expires. +#[tokio::test] +async fn test_hold_timer_expires() -> Result<()> { + // hold_time = 3 s → server expects a KEEPALIVE within 3 s. + let server = TestServer::start_with_hold_time(65000, 3).await?; + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 3).await?; + + // Don't send any keepalives. The server will send keepalives until its own + // hold timer fires and it sends a NOTIFICATION. Drain messages until we + // see it (or the outer timeout fires). + let notif = tokio::time::timeout(Duration::from_secs(6), async { + loop { + match peer.recv().await? { + Message::Notification(n) => return Ok::<_, eyre::Error>(n), + Message::KeepAlive => {} // server keepalive — keep waiting + other => eyre::bail!("Unexpected message: {:?}", other), + } + } + }) + .await??; + assert!( + matches!(notif.error, BgpError::HoldTimer), + "Expected HoldTimer NOTIFICATION, got {:?}", + notif + ); + Ok(()) +} + +/// A peer-sent NOTIFICATION is handled gracefully (session closes cleanly). +#[tokio::test] +async fn test_peer_sends_notification() -> Result<()> { + let server = TestServer::start(65000).await?; + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90) + .await?; + + peer.send(&Message::Notification(NotificationMessage { + error: BgpError::Cease(CeaseSubcode::AdminShutdown), + data: vec![], + })) + .await?; + + // Server should close the connection cleanly. + tokio::time::sleep(Duration::from_millis(100)).await; + assert!(peer.is_closed().await); + Ok(()) +} + +// --------------------------------------------------------------------------- +// UPDATE +// --------------------------------------------------------------------------- + +/// An UPDATE message is accepted without crashing the server. +#[tokio::test] +async fn test_server_accepts_update() -> Result<()> { + let server = TestServer::start(65000).await?; + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90) + .await?; + + let update = Message::Update(UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![], + // 10.0.0.0/8 — one byte of prefix for a /8. + announced_routes: vec![IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?], + }); + peer.send(&update).await?; + + // Server should stay up — we can still exchange a keepalive. + peer.send(&Message::KeepAlive).await?; + Ok(()) +} + +/// An IPv6 UPDATE via MP_REACH_NLRI is accepted without crashing the server. +#[tokio::test] +async fn test_server_accepts_ipv6_update() -> Result<()> { + let server = TestServer::start(65000).await?; + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90) + .await?; + + // 2001:db8::/32 + let prefix = IpPrefix::new(AddressFamilyId::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?; + let update = Message::Update(UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![PathAttribute::MpReachNlri(MpReachNlriPathAttribute { + afi: AddressFamilyId::Ipv6, + safi: SubsequentAfi::Unicast, + next_hop: NlriNextHop::Ipv6("2001:db8::1".parse().unwrap()), + prefixes: vec![prefix], + })], + announced_routes: vec![], + }); + peer.send(&update).await?; + + // Server should stay up — we can still exchange a keepalive. + peer.send(&Message::KeepAlive).await?; + Ok(()) +} + +// --------------------------------------------------------------------------- +// Adj-RIB-In introspection +// --------------------------------------------------------------------------- + +/// IPv4 prefix announced via the legacy NLRI field appears in the RIB. +#[tokio::test] +async fn test_rib_records_ipv4_announce() -> Result<()> { + let server = TestServer::start(65000).await?; + let mut events = server.adj_rib_in.subscribe(); + + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90).await?; + + let prefix = IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?; + peer.send(&Message::Update(UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![], + announced_routes: vec![prefix.clone()], + })) + .await?; + + let event = tokio::time::timeout(Duration::from_secs(1), events.recv()).await??; + assert!( + matches!(&event, RibEvent::Announced { prefix: p, .. } if p == &prefix), + "Expected Announced for 10.0.0.0/8, got {event:?}", + ); + + let snapshot = server.adj_rib_in.snapshot().await; + assert!( + snapshot.iter().any(|e| e.prefix == prefix), + "10.0.0.0/8 missing from RIB snapshot", + ); + Ok(()) +} + +/// Withdrawn IPv4 prefix is removed from the RIB. +#[tokio::test] +async fn test_rib_records_ipv4_withdraw() -> Result<()> { + let server = TestServer::start(65000).await?; + let mut events = server.adj_rib_in.subscribe(); + + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90).await?; + + let prefix = IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?; + + // Announce first. + peer.send(&Message::Update(UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![], + announced_routes: vec![prefix.clone()], + })) + .await?; + tokio::time::timeout(Duration::from_secs(1), events.recv()).await??; + + // Now withdraw. + peer.send(&Message::Update(UpdateMessage { + withdrawn_routes: vec![prefix.clone()], + path_attributes: vec![], + announced_routes: vec![], + })) + .await?; + + let event = tokio::time::timeout(Duration::from_secs(1), events.recv()).await??; + assert!( + matches!(&event, RibEvent::Withdrawn { prefix: p, .. } if p == &prefix), + "Expected Withdrawn for 10.0.0.0/8, got {event:?}", + ); + + let snapshot = server.adj_rib_in.snapshot().await; + assert!( + !snapshot.iter().any(|e| e.prefix == prefix), + "10.0.0.0/8 should have been removed from RIB snapshot", + ); + Ok(()) +} + +/// IPv6 prefix announced via MP_REACH_NLRI appears in the RIB. +#[tokio::test] +async fn test_rib_records_ipv6_announce() -> Result<()> { + let server = TestServer::start(65000).await?; + let mut events = server.adj_rib_in.subscribe(); + + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90).await?; + + let prefix = IpPrefix::new(AddressFamilyId::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?; + peer.send(&Message::Update(UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![PathAttribute::MpReachNlri(MpReachNlriPathAttribute { + afi: AddressFamilyId::Ipv6, + safi: SubsequentAfi::Unicast, + next_hop: NlriNextHop::Ipv6("2001:db8::1".parse().unwrap()), + prefixes: vec![prefix.clone()], + })], + announced_routes: vec![], + })) + .await?; + + let event = tokio::time::timeout(Duration::from_secs(1), events.recv()).await??; + assert!( + matches!(&event, RibEvent::Announced { prefix: p, .. } if p == &prefix), + "Expected Announced for 2001:db8::/32, got {event:?}", + ); + + let snapshot = server.adj_rib_in.snapshot().await; + assert!( + snapshot.iter().any(|e| e.prefix == prefix), + "2001:db8::/32 missing from RIB snapshot", + ); + Ok(()) +} + +/// IPv6 prefix withdrawn via MP_UNREACH_NLRI is removed from the RIB. +#[tokio::test] +async fn test_rib_records_ipv6_withdraw() -> Result<()> { + let server = TestServer::start(65000).await?; + let mut events = server.adj_rib_in.subscribe(); + + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90).await?; + + let prefix = IpPrefix::new(AddressFamilyId::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?; + + // Announce first. + peer.send(&Message::Update(UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![PathAttribute::MpReachNlri(MpReachNlriPathAttribute { + afi: AddressFamilyId::Ipv6, + safi: SubsequentAfi::Unicast, + next_hop: NlriNextHop::Ipv6("2001:db8::1".parse().unwrap()), + prefixes: vec![prefix.clone()], + })], + announced_routes: vec![], + })) + .await?; + tokio::time::timeout(Duration::from_secs(1), events.recv()).await??; + + // Now withdraw via MP_UNREACH_NLRI. + peer.send(&Message::Update(UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute { + afi: AddressFamilyId::Ipv6, + safi: SubsequentAfi::Unicast, + prefixes: vec![prefix.clone()], + })], + announced_routes: vec![], + })) + .await?; + + let event = tokio::time::timeout(Duration::from_secs(1), events.recv()).await??; + assert!( + matches!(&event, RibEvent::Withdrawn { prefix: p, .. } if p == &prefix), + "Expected Withdrawn for 2001:db8::/32, got {event:?}", + ); + + let snapshot = server.adj_rib_in.snapshot().await; + assert!( + !snapshot.iter().any(|e| e.prefix == prefix), + "2001:db8::/32 should have been removed from RIB snapshot", + ); + Ok(()) +} diff --git a/crates/packet/src/message.rs b/crates/packet/src/message.rs index d009cf5..58daa8d 100644 --- a/crates/packet/src/message.rs +++ b/crates/packet/src/message.rs @@ -1446,7 +1446,13 @@ impl NotificationMessage { let error = BgpError::from_code_subcode(error_code, error_subcode) .map_err(|e| Failure(BgpParserError::Eyre(e.into())))?; - Ok((&[], Self { error, data: buf.to_vec() })) + Ok(( + &[], + Self { + error, + data: buf.to_vec(), + }, + )) } pub fn to_wire(&self, _ctx: &ParserContext, buf: &mut BytesMut) -> Result<()> { diff --git a/crates/packet/src/open.rs b/crates/packet/src/open.rs index 64e1a46..0d33fbd 100644 --- a/crates/packet/src/open.rs +++ b/crates/packet/src/open.rs @@ -314,7 +314,7 @@ impl Capability { let (buf, cap_buf) = nom::bytes::take(len).parse(buf)?; macro_rules! require_empty { ($buf:expr, $cap_type:expr) => { - if !buf.is_empty() { + if !$buf.is_empty() { return Err(nom::Err::Failure(BgpParserError::Eyre(eyre!( "Unexpected leftover bytes {:x?} while parsing {} capability", $buf, @@ -456,7 +456,7 @@ impl Capability { v if v == CapabilityDiscriminants::FourByteAsn as u8 => { let (cap_buf, asn) = be_u32(cap_buf)?; require_empty!(cap_buf, "FourByteAsn"); - Ok((cap_buf, Self::FourByteAsn { asn })) + Ok((buf, Self::FourByteAsn { asn })) } v if v == CapabilityDiscriminants::AddPath as u8 => { let (cap_buf, afi) = be_u16(cap_buf)?; diff --git a/crates/rib/Cargo.toml b/crates/rib/Cargo.toml new file mode 100644 index 0000000..1b4a644 --- /dev/null +++ b/crates/rib/Cargo.toml @@ -0,0 +1,9 @@ +[package] +name = "bgp-rib" +edition.workspace = true + +[dependencies] +bgp-packet.workspace = true +bgp-primitives.workspace = true +tokio = { workspace = true, features = ["full"] } +tracing.workspace = true diff --git a/crates/rib/src/adj_rib_in.rs b/crates/rib/src/adj_rib_in.rs new file mode 100644 index 0000000..4f7cf53 --- /dev/null +++ b/crates/rib/src/adj_rib_in.rs @@ -0,0 +1,96 @@ +use std::collections::HashMap; + +use bgp_primitives::ip_prefix::IpPrefix; +use tokio::sync::{broadcast, mpsc, oneshot}; +use tracing::warn; + +use crate::entry::{PeerId, RibEntry, SnapshotEntry}; +use crate::event::RibEvent; + +const EVENT_CHANNEL_CAPACITY: usize = 1024; +const CMD_CHANNEL_CAPACITY: usize = 256; + +enum RibCommand { + Announce { peer: PeerId, prefix: IpPrefix, entry: RibEntry }, + Withdraw { peer: PeerId, prefix: IpPrefix }, + Snapshot { reply: oneshot::Sender> }, +} + +/// Handle to the Adj-RIB-In actor. Cheap to clone — all clones share the same +/// backing task. The actor is stopped when all handles are dropped. +#[derive(Clone)] +pub struct AdjRibIn { + tx: mpsc::Sender, + events: broadcast::Sender, +} + +impl AdjRibIn { + /// Spawns the backing actor and returns a handle. + pub fn spawn() -> Self { + let (tx, rx) = mpsc::channel(CMD_CHANNEL_CAPACITY); + let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY); + let events_clone = events.clone(); + tokio::spawn(run_actor(rx, events_clone)); + Self { tx, events } + } + + pub async fn announce(&self, peer: PeerId, prefix: IpPrefix, entry: RibEntry) { + let _ = self.tx.send(RibCommand::Announce { peer, prefix, entry }).await; + } + + pub async fn withdraw(&self, peer: PeerId, prefix: IpPrefix) { + let _ = self.tx.send(RibCommand::Withdraw { peer, prefix }).await; + } + + /// Returns a point-in-time snapshot of all routes in the RIB. + pub async fn snapshot(&self) -> Vec { + let (reply_tx, reply_rx) = oneshot::channel(); + let _ = self.tx.send(RibCommand::Snapshot { reply: reply_tx }).await; + reply_rx.await.unwrap_or_default() + } + + /// Subscribe to a stream of RIB mutation events. + /// + /// The receiver will see every [`RibEvent`] from the point of subscription. + /// If the consumer falls behind by more than [`EVENT_CHANNEL_CAPACITY`] + /// events it will receive a [`broadcast::error::RecvError::Lagged`] error. + pub fn subscribe(&self) -> broadcast::Receiver { + self.events.subscribe() + } +} + +async fn run_actor(mut rx: mpsc::Receiver, events: broadcast::Sender) { + let mut table: HashMap<(PeerId, IpPrefix), RibEntry> = HashMap::new(); + + while let Some(cmd) = rx.recv().await { + match cmd { + RibCommand::Announce { peer, prefix, entry } => { + table.insert((peer, prefix.clone()), entry.clone()); + if let Err(e) = events.send(RibEvent::Announced { peer, prefix, entry }) { + // No active subscribers — not an error. + let _ = e; + } + } + RibCommand::Withdraw { peer, prefix } => { + if table.remove(&(peer, prefix.clone())).is_some() { + if let Err(e) = events.send(RibEvent::Withdrawn { peer, prefix }) { + let _ = e; + } + } + } + RibCommand::Snapshot { reply } => { + let snap = table + .iter() + .map(|((peer, prefix), entry)| SnapshotEntry { + peer: *peer, + prefix: prefix.clone(), + entry: entry.clone(), + }) + .collect(); + if reply.send(snap).is_err() { + warn!("Snapshot reply channel closed before send"); + } + } + } + } +} diff --git a/crates/rib/src/entry.rs b/crates/rib/src/entry.rs new file mode 100644 index 0000000..bdf278e --- /dev/null +++ b/crates/rib/src/entry.rs @@ -0,0 +1,39 @@ +use std::net::SocketAddr; +use std::time::SystemTime; + +use bgp_packet::message::PathAttribute; +use bgp_primitives::ip_prefix::IpPrefix; + +/// Identifies a BGP peer by its TCP address and negotiated ASN. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct PeerId { + pub addr: SocketAddr, + pub asn: u32, +} + +impl std::fmt::Display for PeerId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}(AS{})", self.addr, self.asn) + } +} + +/// A single route stored in the Adj-RIB-In. +#[derive(Debug, Clone)] +pub struct RibEntry { + pub path_attributes: Vec, + pub received_at: SystemTime, +} + +impl RibEntry { + pub fn new(path_attributes: Vec) -> Self { + Self { path_attributes, received_at: SystemTime::now() } + } +} + +/// A flattened row returned by [`crate::AdjRibIn::snapshot`]. +#[derive(Debug, Clone)] +pub struct SnapshotEntry { + pub peer: PeerId, + pub prefix: IpPrefix, + pub entry: RibEntry, +} diff --git a/crates/rib/src/event.rs b/crates/rib/src/event.rs new file mode 100644 index 0000000..868afdd --- /dev/null +++ b/crates/rib/src/event.rs @@ -0,0 +1,9 @@ +use bgp_primitives::ip_prefix::IpPrefix; + +use crate::entry::{PeerId, RibEntry}; + +#[derive(Debug, Clone)] +pub enum RibEvent { + Announced { peer: PeerId, prefix: IpPrefix, entry: RibEntry }, + Withdrawn { peer: PeerId, prefix: IpPrefix }, +} diff --git a/crates/rib/src/lib.rs b/crates/rib/src/lib.rs new file mode 100644 index 0000000..8bb39e6 --- /dev/null +++ b/crates/rib/src/lib.rs @@ -0,0 +1,7 @@ +mod adj_rib_in; +mod entry; +mod event; + +pub use adj_rib_in::AdjRibIn; +pub use entry::{PeerId, RibEntry, SnapshotEntry}; +pub use event::RibEvent; diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 1852472..e1a5686 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -3,7 +3,12 @@ name = "bgp-server" edition.workspace = true [dependencies] +bgp-packet.workspace = true +bgp-primitives.workspace = true +bgp-rib.workspace = true +bytes.workspace = true clap.workspace = true +futures.workspace = true eyre.workspace = true prometheus.workspace = true tokio.workspace = true diff --git a/crates/server/proto/config.proto b/crates/server/proto/config.proto index 25c46e3..3210367 100644 --- a/crates/server/proto/config.proto +++ b/crates/server/proto/config.proto @@ -4,29 +4,41 @@ package bgp.server; // Top level BGP Server configuration. message BgpConfig { - // The IPv4 address used as the router ID in BGP. + // The router ID, written as a dotted-quad IPv4 address (e.g. "192.0.2.1"). + // Required even on IPv6-only routers per RFC 4271. string router_id = 1; - // The local AS Number to use when communicating with peers. + + // The local 4-byte AS number (RFC 6793). uint32 local_asn = 2; - // Endpoints to listen on. + // Endpoints to accept incoming BGP connections on. + // Defaults to 0.0.0.0:179 and [::]:179 when empty. repeated TcpEndpoint listen_endpoints = 3; // Peers configured on this server. repeated Peer peers = 4; } -// Represents an IP address and port that can be used for listening or connecting. +// An IP address and port pair. message TcpEndpoint { - // Address to listen on + // IP address (V4 or V6). string address = 1; - // Port to listen on. + // TCP port. Defaults to 179 for BGP when unset. uint32 port = 2; } // Configuration of a BGP peer. message Peer { + // Human-readable name for logging and diagnostics. string name = 1; - TcpEndpoint address = 2; + + // Address and port to connect to for this peer. + TcpEndpoint endpoint = 2; + + // The peer's 4-byte AS number (RFC 6793). uint32 remote_asn = 3; + + // Hold time to advertise in the OPEN message, in seconds (RFC 4271 §4.2). + // Must be 0 or >= 3. Defaults to 90 when unset. + uint32 hold_time_seconds = 4; } diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs new file mode 100644 index 0000000..317ea7a --- /dev/null +++ b/crates/server/src/config.rs @@ -0,0 +1,94 @@ +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::str::FromStr; + +use eyre::{Result, bail, eyre}; + +use crate::proto::config::BgpConfig; + +const DEFAULT_BGP_PORT: u16 = 179; +const DEFAULT_HOLD_TIME_SECS: u16 = 90; + +/// Validated, Rust-native server configuration derived from the proto config. +pub struct Config { + pub router_id: Ipv4Addr, + pub local_asn: u32, + /// Default hold time advertised to peers that have no per-peer override. + pub hold_time_secs: u16, + pub listen_endpoints: Vec, + pub peers: Vec, +} + +pub struct PeerConfig { + pub name: String, + pub endpoint: SocketAddr, + pub remote_asn: u32, + pub hold_time_secs: u16, +} + +impl Config { + pub fn from_proto(proto: BgpConfig) -> Result { + let router_id = Ipv4Addr::from_str(&proto.router_id) + .map_err(|_| eyre!("Invalid router_id {:?}: must be a dotted-quad IPv4 address", proto.router_id))?; + + if proto.local_asn == 0 { + bail!("local_asn must be non-zero"); + } + + let listen_endpoints = if proto.listen_endpoints.is_empty() { + vec![ + SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), DEFAULT_BGP_PORT), + SocketAddr::new(IpAddr::V6(std::net::Ipv6Addr::UNSPECIFIED), DEFAULT_BGP_PORT), + ] + } else { + proto.listen_endpoints + .into_iter() + .map(|ep| parse_endpoint(&ep.address, ep.port)) + .collect::>()? + }; + + let peers = proto.peers + .into_iter() + .map(|p| { + let ep = p.endpoint.as_ref() + .ok_or_else(|| eyre!("Peer {:?} is missing endpoint", p.name))?; + let endpoint = parse_endpoint(&ep.address, ep.port)?; + + if p.remote_asn == 0 { + bail!("Peer {:?} remote_asn must be non-zero", p.name); + } + + let hold_time_secs = if p.hold_time_seconds == 0 { + DEFAULT_HOLD_TIME_SECS + } else if p.hold_time_seconds < 3 { + bail!("Peer {:?} hold_time_seconds must be 0 or >= 3", p.name); + } else { + p.hold_time_seconds as u16 + }; + + Ok(PeerConfig { + name: p.name, + endpoint, + remote_asn: p.remote_asn, + hold_time_secs, + }) + }) + .collect::>()?; + + Ok(Self { router_id, local_asn: proto.local_asn, hold_time_secs: DEFAULT_HOLD_TIME_SECS, listen_endpoints, peers }) + } +} + +fn parse_endpoint(address: &str, port: u32) -> Result { + let port = if port == 0 { + DEFAULT_BGP_PORT + } else if port > u16::MAX as u32 { + bail!("Port {port} is out of range"); + } else { + port as u16 + }; + + let ip = IpAddr::from_str(address) + .map_err(|_| eyre!("Invalid IP address {address:?}"))?; + + Ok(SocketAddr::new(ip, port)) +} diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs new file mode 100644 index 0000000..fe9690c --- /dev/null +++ b/crates/server/src/connection.rs @@ -0,0 +1,17 @@ +use std::net::SocketAddr; +use std::sync::Arc; + +use tokio::net::TcpStream; +use tracing::warn; + +use crate::fsm::{SessionConfig, run_session}; + +pub async fn handle_connection( + stream: TcpStream, + peer_addr: SocketAddr, + config: Arc, +) { + if let Err(e) = run_session(stream, peer_addr, (*config).clone()).await { + warn!(%peer_addr, err = %e, "BGP session terminated with error"); + } +} diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs new file mode 100644 index 0000000..7a9ba48 --- /dev/null +++ b/crates/server/src/fsm.rs @@ -0,0 +1,389 @@ +use std::net::{Ipv4Addr, SocketAddr}; +use std::time::Duration; + +use bgp_packet::errors::{BgpError, FsmSubcode, OpenMsgSubcode}; +use bgp_packet::message::{BGP_HEADER_LEN, Message, NotificationMessage, PathAttribute}; +use bgp_packet::open::{Capability, OpenMessage, OpenOption}; +use bgp_primitives::constants::{AddressFamilyId, SubsequentAfi}; +use bgp_primitives::parser::ParserContext; +use bgp_rib::{AdjRibIn, PeerId, RibEntry}; +use bytes::BytesMut; +use eyre::{Result, bail, eyre}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::time::{Instant, interval_at, sleep_until}; +use tracing::{info, warn}; + +use bgp_packet::message::{AS_TRANS, BGP4_VERSION}; + +/// Minimal per-session config extracted from the server config. +#[derive(Clone)] +pub struct SessionConfig { + pub router_id: Ipv4Addr, + pub local_asn: u32, + pub hold_time_secs: u16, + pub adj_rib_in: AdjRibIn, +} + +/// Drive the BGP FSM for a single accepted TCP connection. +/// +/// States visited: OpenSent → OpenConfirm → Established. +pub async fn run_session( + stream: TcpStream, + peer_addr: SocketAddr, + config: SessionConfig, +) -> Result<()> { + let (mut reader, mut writer) = stream.into_split(); + + // --- OpenSent ----------------------------------------------------------- + // Send our OPEN immediately after the TCP connection is established. + let our_open = build_open(&config); + send_message(&mut writer, &our_open, &ParserContext::default()).await?; + + // RFC 4271 §8.1: use a large initial hold timer (4 minutes) before we + // know the negotiated value. + let peer_msg = recv_with_timeout( + &mut reader, + &ParserContext::default(), + Duration::from_secs(240), + ) + .await; + + let peer_open = match peer_msg { + Ok(Message::Open(o)) => o, + Ok(other) => { + send_notification( + &mut writer, + BgpError::FsmError(FsmSubcode::UnexpectedOpenSent), + &ParserContext::default(), + ) + .await; + bail!("Expected OPEN in OpenSent state, got {:?}", other); + } + Err(e) => { + send_notification( + &mut writer, + BgpError::FsmError(FsmSubcode::UnexpectedOpenSent), + &ParserContext::default(), + ) + .await; + return Err(e); + } + }; + + if let Err(bgp_err) = validate_open(&peer_open) { + send_notification(&mut writer, bgp_err, &ParserContext::default()).await; + bail!("Peer {} sent invalid OPEN", peer_addr); + } + + let negotiated_hold_time = config.hold_time_secs.min(peer_open.hold_time); + let peer_four_byte_asn = peer_open.options.iter().any(|o| { + matches!(o, OpenOption::Capabilities(caps) + if caps.iter().any(|c| matches!(c, Capability::FourByteAsn { .. }))) + }); + let ctx = ParserContext { + four_octet_asn: Some(peer_four_byte_asn), + address_family: None, + }; + + let peer_asn = if peer_four_byte_asn { + peer_open + .options + .iter() + .find_map(|o| match o { + OpenOption::Capabilities(caps) => caps.iter().find_map(|c| match c { + Capability::FourByteAsn { asn } => Some(*asn), + _ => None, + }), + }) + .unwrap_or(peer_open.asn as u32) + } else { + peer_open.asn as u32 + }; + + info!( + %peer_addr, + peer_asn, + negotiated_hold_time, + "OPEN received, transitioning to OpenConfirm" + ); + + // --- OpenConfirm -------------------------------------------------------- + send_message(&mut writer, &Message::KeepAlive, &ctx).await?; + + let hold_duration = Duration::from_secs(negotiated_hold_time as u64); + + match recv_with_timeout(&mut reader, &ctx, hold_duration).await { + Ok(Message::KeepAlive) => {} + Ok(other) => { + send_notification( + &mut writer, + BgpError::FsmError(FsmSubcode::UnexpectedOpenConfirm), + &ctx, + ) + .await; + bail!("Expected KEEPALIVE in OpenConfirm state, got {:?}", other); + } + Err(e) => { + send_notification(&mut writer, BgpError::HoldTimer, &ctx).await; + return Err(e); + } + } + + info!(%peer_addr, peer_asn, "Session established"); + + let peer_id = PeerId { addr: peer_addr, asn: peer_asn }; + + // --- Established -------------------------------------------------------- + established_loop( + &mut reader, + &mut writer, + &ctx, + negotiated_hold_time, + peer_id, + &config.adj_rib_in, + ) + .await +} + +async fn established_loop( + reader: &mut OwnedReadHalf, + writer: &mut OwnedWriteHalf, + ctx: &ParserContext, + hold_time_secs: u16, + peer_id: PeerId, + adj_rib_in: &AdjRibIn, +) -> 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); + + // interval_at avoids firing immediately; first keepalive is sent after one interval. + let mut keepalive_timer = interval_at(Instant::now() + keepalive_interval, keepalive_interval); + + // sleep_until must be pinned so we can call .as_mut().reset() inside select!. + let hold_timer = sleep_until(Instant::now() + hold_duration); + tokio::pin!(hold_timer); + + loop { + tokio::select! { + result = read_message(reader, ctx) => { + match result? { + Message::KeepAlive => { + hold_timer.as_mut().reset(Instant::now() + hold_duration); + } + 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 => {} + } + } + _ = keepalive_timer.tick() => { + send_message(writer, &Message::KeepAlive, ctx).await?; + } + _ = &mut hold_timer => { + warn!(peer = %peer_id, "Hold timer expired"); + send_notification(writer, BgpError::HoldTimer, ctx).await; + bail!("Hold timer expired for peer {}", peer_id); + } + } + } +} + +/// 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 +// --------------------------------------------------------------------------- + +async fn read_message(reader: &mut OwnedReadHalf, ctx: &ParserContext) -> Result { + let mut header = [0u8; BGP_HEADER_LEN as usize]; + reader.read_exact(&mut header).await?; + + let total_len = u16::from_be_bytes([header[16], header[17]]) as usize; + if total_len < BGP_HEADER_LEN as usize { + bail!("BGP message length {total_len} below minimum"); + } + + let mut buf = vec![0u8; total_len]; + buf[..BGP_HEADER_LEN as usize].copy_from_slice(&header); + reader + .read_exact(&mut buf[BGP_HEADER_LEN as usize..]) + .await?; + + let (_, msg) = + Message::from_wire(ctx, &buf).map_err(|e| eyre!("Failed to parse BGP message: {e:?}"))?; + Ok(msg) +} + +async fn recv_with_timeout( + reader: &mut OwnedReadHalf, + ctx: &ParserContext, + timeout: Duration, +) -> Result { + tokio::time::timeout(timeout, read_message(reader, ctx)) + .await + .map_err(|_| eyre!("Timed out waiting for BGP message"))? +} + +async fn send_message( + writer: &mut OwnedWriteHalf, + msg: &Message, + ctx: &ParserContext, +) -> Result<()> { + let mut buf = BytesMut::new(); + msg.to_wire(ctx, &mut buf)?; + writer.write_all(&buf).await?; + Ok(()) +} + +/// Best-effort NOTIFICATION send — errors are logged and swallowed since we +/// are already tearing down the session. +async fn send_notification(writer: &mut OwnedWriteHalf, error: BgpError, ctx: &ParserContext) { + let msg = Message::Notification(NotificationMessage { + error, + data: vec![], + }); + if let Err(e) = send_message(writer, &msg, ctx).await { + warn!("Failed to send NOTIFICATION: {e}"); + } +} + +// --------------------------------------------------------------------------- +// RIB helpers +// --------------------------------------------------------------------------- + +/// Applies a received UPDATE to the Adj-RIB-In. +/// +/// Handles both the legacy IPv4 NLRI fields and the MP_REACH/MP_UNREACH path +/// attributes used for all other address families (RFC 2858). +async fn apply_update( + rib: &AdjRibIn, + peer: PeerId, + path_attributes: &[PathAttribute], + withdrawn: &[bgp_primitives::ip_prefix::IpPrefix], + announced: &[bgp_primitives::ip_prefix::IpPrefix], +) { + // Legacy IPv4 fields in the UPDATE body. + for prefix in withdrawn { + rib.withdraw(peer, prefix.clone()).await; + } + for prefix in announced { + rib.announce(peer, prefix.clone(), RibEntry::new(path_attributes.to_vec())).await; + } + + // Multiprotocol attributes carry NLRI for all other AFIs. + for attr in path_attributes { + match attr { + PathAttribute::MpReachNlri(mp) => { + for prefix in &mp.prefixes { + rib.announce(peer, prefix.clone(), RibEntry::new(path_attributes.to_vec())).await; + } + } + PathAttribute::MpUnreachNlri(mp) => { + for prefix in &mp.prefixes { + rib.withdraw(peer, prefix.clone()).await; + } + } + _ => {} + } + } +} + +// --------------------------------------------------------------------------- +// OPEN construction and validation +// --------------------------------------------------------------------------- + +fn build_open(config: &SessionConfig) -> Message { + // Use AS_TRANS in the 2-byte field when the ASN doesn't fit; the real ASN + // is carried in the FourByteAsn capability (RFC 6793). + let asn_field = if config.local_asn > u16::MAX as u32 { + AS_TRANS + } else { + config.local_asn as u16 + }; + + Message::Open(OpenMessage { + version: BGP4_VERSION, + asn: asn_field, + hold_time: config.hold_time_secs, + identifier: config.router_id, + options: vec![OpenOption::Capabilities(vec![ + Capability::FourByteAsn { + asn: config.local_asn, + }, + Capability::MultiProtocol { + afi: AddressFamilyId::Ipv4, + safi: SubsequentAfi::Unicast, + }, + Capability::RouteRefresh, + ])], + }) +} + +fn validate_open(open: &OpenMessage) -> Result<(), BgpError> { + if open.version != BGP4_VERSION { + return Err(BgpError::OpenMsg(OpenMsgSubcode::UnsupportedVersion)); + } + if open.hold_time != 0 && open.hold_time < 3 { + return Err(BgpError::OpenMsg(OpenMsgSubcode::UnacceptableHoldTime)); + } + Ok(()) +} diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 62b1359..e88955b 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -1,3 +1,7 @@ +pub mod config; +pub mod connection; +pub mod fsm; +pub mod server; pub mod tracing; pub mod proto { diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index 7e5500e..61f9df0 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -1,26 +1,29 @@ -use bgp_server::proto; +use bgp_server::config::Config; +use bgp_server::proto::config::BgpConfig; +use bgp_server::server::BgpServerBuilder; +use bgp_server::tracing::TracingArgs; use clap::Parser; use eyre::{Context, Result}; -use tracing::info; - -use crate::proto::config::BgpConfig; #[derive(Parser, Debug)] struct ServerArgs { - /// Path to the configuration file. + /// Path to the textproto configuration file. config: String, + + #[command(flatten)] + tracing: TracingArgs, } #[tokio::main] async fn main() -> Result<()> { - info!("Starting bgp server"); - let args = ServerArgs::parse(); + args.tracing.install()?; + let config_string = - std::fs::read_to_string(args.config).wrap_err("Failed to open config file")?; - let config = protobuf::text_format::parse_from_str::(&config_string)?; + std::fs::read_to_string(&args.config).wrap_err("Failed to open config file")?; + let proto_config = protobuf::text_format::parse_from_str::(&config_string) + .wrap_err("Failed to parse config file")?; + let config = Config::from_proto(proto_config).wrap_err("Invalid configuration")?; - info!("Read config file: {}", config); - - Ok(()) + BgpServerBuilder::new(config).run().await } diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs new file mode 100644 index 0000000..69f791c --- /dev/null +++ b/crates/server/src/server.rs @@ -0,0 +1,114 @@ +use std::io; +use std::net::SocketAddr; +use std::sync::Arc; + +use eyre::{Result, WrapErr}; +use futures::stream::{FuturesUnordered, StreamExt}; +use tokio::net::{TcpListener, TcpStream}; +use tracing::{error, info}; + +use bgp_rib::AdjRibIn; + +use crate::config::Config; +use crate::connection::handle_connection; +use crate::fsm::SessionConfig; + +pub struct BgpServerBuilder { + config: Config, +} + +/// A `BgpServerBuilder` that has bound all its listeners and is ready to accept. +/// Call [`BgpServer::run`] to start the accept loop. +pub struct BgpServer { + config: Config, + listeners: Vec<(SocketAddr, TcpListener)>, + /// 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, +} + +impl BgpServerBuilder { + pub fn new(config: Config) -> Self { + Self { config } + } + + /// Binds all configured listen endpoints and returns a [`BgpServer`]. + /// Separating bind from run lets callers (e.g. tests) inspect the actual + /// bound addresses before the accept loop starts. + pub async fn bind(self) -> Result { + let mut listeners = Vec::with_capacity(self.config.listen_endpoints.len()); + for &addr in &self.config.listen_endpoints { + let listener = TcpListener::bind(addr) + .await + .wrap_err_with(|| format!("Failed to bind listener on {addr}"))?; + let bound_addr = listener.local_addr()?; + info!(%bound_addr, "Listening for BGP connections"); + listeners.push((bound_addr, listener)); + } + Ok(BgpServer { + config: self.config, + listeners, + adj_rib_in: AdjRibIn::spawn(), + }) + } + + /// Convenience method: bind then run immediately. + pub async fn run(self) -> Result<()> { + self.bind().await?.run().await + } +} + +impl BgpServer { + /// The addresses actually bound (useful when the config uses port 0). + pub fn local_addrs(&self) -> Vec { + self.listeners.iter().map(|(addr, _)| *addr).collect() + } + + pub async fn run(self) -> Result<()> { + info!( + router_id = %self.config.router_id, + local_asn = self.config.local_asn, + peers = self.config.peers.len(), + "BGP server started" + ); + + let session_config = Arc::new(SessionConfig { + router_id: self.config.router_id, + local_asn: self.config.local_asn, + hold_time_secs: self.config.hold_time_secs, + adj_rib_in: self.adj_rib_in.clone(), + }); + + let mut accepting: FuturesUnordered<_> = self + .listeners + .into_iter() + .map(|(addr, listener)| accept_one(addr, listener)) + .collect(); + + while let Some((listen_addr, result, listener)) = accepting.next().await { + match result { + Ok((stream, peer_addr)) => { + tokio::spawn(handle_connection( + stream, + peer_addr, + Arc::clone(&session_config), + )); + } + Err(e) => { + error!(%listen_addr, err = %e, "Accept error"); + } + } + accepting.push(accept_one(listen_addr, listener)); + } + + Ok(()) + } +} + +async fn accept_one( + addr: SocketAddr, + listener: TcpListener, +) -> (SocketAddr, io::Result<(TcpStream, SocketAddr)>, TcpListener) { + let result = listener.accept().await; + (addr, result, listener) +} diff --git a/crates/server/src/tracing.rs b/crates/server/src/tracing.rs index ad6dbfb..2c22065 100644 --- a/crates/server/src/tracing.rs +++ b/crates/server/src/tracing.rs @@ -25,6 +25,7 @@ impl TracingArgs { }, )?; + tracing_subscriber::fmt().with_env_filter(filter).init(); Ok(()) } } diff --git a/docs/RFCs.md b/docs/RFCs.md index 8c8f5ad..b37f359 100644 --- a/docs/RFCs.md +++ b/docs/RFCs.md @@ -1,9 +1,164 @@ # RFCs -This document refers to the relevant RFCs for BGP and associated networking protocol semantics that we implement in this project. +Tracks which RFCs are implemented, partially implemented, or only represented +in the wire format layer. -## RFC 4271 - Border Gateway protocol +**Legend** +- ✅ Implemented — functional end-to-end +- 🟡 Wire format only — parsed and serialised correctly, but not yet acted on +- ❌ Not implemented +--- + +## Core BGP + +### RFC 4271 — A Border Gateway Protocol 4 (BGP-4) https://datatracker.ietf.org/doc/html/rfc4271 -This is the main specification document for the Border Gateway Protocol, and defines the core protocol semantics and wire format. +The base protocol specification. + +| Section | Description | Status | +|---------|-------------|--------| +| §4.1 | Message framing (marker, length, type) | ✅ | +| §4.2 | OPEN message | ✅ | +| §4.3 | UPDATE message (path attributes, NLRI, withdrawn routes) | ✅ | +| §4.4 | KEEPALIVE message | ✅ | +| §4.5 | NOTIFICATION message (all error codes and subcodes) | ✅ | +| §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) | ❌ | + +--- + +## Capability and Session Extensions + +### RFC 6793 — BGP Support for Four-Octet Autonomous System (AS) Number Space +https://datatracker.ietf.org/doc/html/rfc6793 + +| Feature | Status | +|---------|--------| +| `AS_TRANS` (23456) sentinel | ✅ | +| `FourByteAsn` capability advertisement and parsing | ✅ | +| 4-byte AS_PATH segment encoding/decoding via `ParserContext` | ✅ | +| 4-byte ASN negotiation in FSM | ✅ | + +### RFC 9072 — Extended Optional Parameters Length for BGP OPEN Message +https://datatracker.ietf.org/doc/html/rfc9072 + +| Feature | Status | +|---------|--------| +| Parsing extended-length OPEN (0xFF sentinel) | ✅ | +| Serialising extended-length OPEN when options exceed 255 bytes | ✅ | + +### RFC 2918 — Route Refresh Capability for BGP-4 +https://datatracker.ietf.org/doc/html/rfc2918 + +| Feature | Status | +|---------|--------| +| `RouteRefresh` capability advertisement | ✅ | +| `ROUTE-REFRESH` message type (5) framing | ✅ | +| AFI/SAFI payload parsing | ❌ | +| Acting on received ROUTE-REFRESH (re-advertising routes) | ❌ | + +### RFC 7313 — Enhanced Route Refresh Capability for BGP-4 +https://datatracker.ietf.org/doc/html/rfc7313 + +| Feature | Status | +|---------|--------| +| `EnhancedRouteRefresh` capability wire format | 🟡 | + +### RFC 4724 — Graceful Restart Mechanism for BGP +https://datatracker.ietf.org/doc/html/rfc4724 + +| Feature | Status | +|---------|--------| +| `GracefulRestart` capability wire format | 🟡 | +| Restart state and forwarding preserved across sessions | ❌ | + +### RFC 9494 — Long-Lived Graceful Restart for BGP +https://datatracker.ietf.org/doc/html/rfc9494 + +| Feature | Status | +|---------|--------| +| `LongLivedGracefulRestart` capability wire format | 🟡 | + +### RFC 8654 — Extended Message Support for BGP +https://datatracker.ietf.org/doc/html/rfc8654 + +| Feature | Status | +|---------|--------| +| `ExtendedMessage` capability wire format | 🟡 | +| Raising `BGP_MAX_MSG_LEN` to 65535 when negotiated | ❌ | + +--- + +## Multiprotocol and NLRI + +### RFC 2858 — Multiprotocol Extensions for BGP-4 +https://datatracker.ietf.org/doc/html/rfc2858 + +| Feature | Status | +|---------|--------| +| `MultiProtocol` capability advertisement (IPv4 unicast) | ✅ | +| `MP_REACH_NLRI` path attribute (parse, serialise) | ✅ | +| `MP_UNREACH_NLRI` path attribute (parse, serialise) | ✅ | +| IPv4 and IPv6 next-hop encoding | ✅ | + +### RFC 8950 — Advertising IPv4 Network Layer Reachability Information (NLRI) with an IPv6 Next Hop +https://datatracker.ietf.org/doc/html/rfc8950 + +| Feature | Status | +|---------|--------| +| `ExtendedNextHop` capability wire format | 🟡 | + +--- + +## Path Attributes and Communities + +### RFC 1997 — BGP Communities Attribute +https://datatracker.ietf.org/doc/html/rfc1997 + +| Feature | Status | +|---------|--------| +| `COMMUNITIES` path attribute (parse, serialise) | ✅ | + +### RFC 4360 — BGP Extended Communities Attribute +https://datatracker.ietf.org/doc/html/rfc4360 + +| Feature | Status | +|---------|--------| +| `EXTENDED_COMMUNITIES` path attribute (parse, serialise) | ✅ | +| AS-specific, IPv4-address-specific, opaque, route-target, route-origin subtypes | ✅ | + +### RFC 8092 — BGP Large Communities Attribute +https://datatracker.ietf.org/doc/html/rfc8092 + +| Feature | Status | +|---------|--------| +| `LARGE_COMMUNITIES` path attribute (parse, serialise) | ✅ | + +--- + +## Other Capabilities (Wire Format Only) + +The following capabilities are parsed and serialised correctly but have no +functional behaviour implemented beyond that. + +| RFC | Capability | Status | +|-----|-----------|--------| +| RFC 5291 | Outbound Route Filtering (ORF) | 🟡 | +| RFC 7911 | ADD-PATH | 🟡 | +| RFC 8205 | BGPsec | 🟡 | + +--- + +## Error Handling + +### RFC 9687 — BGP Send Hold Timer +https://datatracker.ietf.org/doc/html/rfc9687 + +| Feature | Status | +|---------|--------| +| `SendHoldTimer` error code (8) defined | ✅ | +| Send hold timer enforcement | ❌ |