diff --git a/crates/integration-tests/src/test_server.rs b/crates/integration-tests/src/test_server.rs index 6181a73..de48c1b 100644 --- a/crates/integration-tests/src/test_server.rs +++ b/crates/integration-tests/src/test_server.rs @@ -58,6 +58,7 @@ impl TestServer { listen_endpoints: vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)], peers: vec![], connect_retry: std::time::Duration::from_secs(5), + collision_detect_established_state: false, local_route_policy: LocalRibPolicy::default(), http_endpoint: None, }; diff --git a/crates/integration-tests/tests/announce.rs b/crates/integration-tests/tests/announce.rs index 222effa..c778a7d 100644 --- a/crates/integration-tests/tests/announce.rs +++ b/crates/integration-tests/tests/announce.rs @@ -28,8 +28,14 @@ fn route(p: &str, attrs: LocalRouteAttrs) -> LocalRoute { } async fn established_peer(server: &TestServer) -> Result { + established_peer_with_id(server, Ipv4Addr::new(10, 0, 0, 1)).await +} + +/// Distinct routers need distinct BGP Identifiers, or the second connection +/// is (correctly) treated as a collision with the first. +async fn established_peer_with_id(server: &TestServer, router_id: Ipv4Addr) -> Result { let mut peer = TestPeer::connect(server.addr).await?; - peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 1), 90).await?; + peer.handshake(PEER_ASN, router_id, 90).await?; Ok(peer) } @@ -143,8 +149,8 @@ async fn test_routes_replayed_on_session_up() -> Result<()> { #[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?; + let mut a = established_peer_with_id(&server, Ipv4Addr::new(10, 0, 0, 1)).await?; + let mut b = established_peer_with_id(&server, Ipv4Addr::new(10, 0, 0, 2)).await?; wait_for_sessions(&server, 2).await?; server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?; diff --git a/crates/integration-tests/tests/collision.rs b/crates/integration-tests/tests/collision.rs new file mode 100644 index 0000000..2b45ed0 --- /dev/null +++ b/crates/integration-tests/tests/collision.rs @@ -0,0 +1,287 @@ +//! Connection collision detection (RFC 4271 §6.8) against a live server. +//! +//! The test server's BGP Identifier is 127.0.0.1. A test peer picks an +//! identifier above or below that to steer which connection must survive. + +use std::net::{Ipv4Addr, SocketAddr}; +use std::time::Duration; + +use bgp_integration_tests::test_peer::TestPeer; +use bgp_integration_tests::test_server::TestServer; +use bgp_packet::errors::{BgpError, CeaseSubcode, OpenMsgSubcode}; +use bgp_packet::message::{ + AsPathAttribute, Message, NextHopPathAttribute, NotificationMessage, OriginPathAttribute, + PathAttribute, UpdateMessage, +}; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rib::{LocalRoute, LocalRouteAttrs}; +use bgp_server::collision::Direction; +use bgp_server::config::PeerConfig; +use eyre::Result; +use tokio::net::TcpListener; + +const LOCAL_ASN: u32 = 65000; +const PEER_ASN: u32 = 65001; +const TIMEOUT: Duration = Duration::from_secs(5); + +/// Above the server's 127.0.0.1. +const HIGH_ID: Ipv4Addr = Ipv4Addr::new(192, 0, 2, 1); +/// Below the server's 127.0.0.1. +const LOW_ID: Ipv4Addr = Ipv4Addr::new(10, 0, 0, 1); + +async fn server_dialing(listener: &TcpListener) -> Result { + let endpoint: SocketAddr = listener.local_addr()?; + TestServer::start_with(LOCAL_ASN, |c| { + c.connect_retry = Duration::from_millis(200); + c.peers = vec![PeerConfig { name: "lab".into(), endpoint, remote_asn: PEER_ASN, hold_time_secs: 90 }]; + }) + .await +} + +/// Exchanges OPENs only, leaving the server's side in OpenConfirm (it has +/// sent its KEEPALIVE and is waiting for ours). +async fn open_only(peer: &mut TestPeer, router_id: Ipv4Addr) -> Result<()> { + peer.send_open(PEER_ASN, router_id, 90).await?; + peer.recv_open().await?; + Ok(()) +} + +/// Finishes the handshake begun by [`open_only`]. +async fn confirm(peer: &mut TestPeer) -> Result<()> { + match tokio::time::timeout(TIMEOUT, peer.recv()).await?? { + Message::KeepAlive => {} + other => eyre::bail!("expected KEEPALIVE, got {other:?}"), + } + peer.send(&Message::KeepAlive).await +} + +/// Reads until a NOTIFICATION arrives, skipping KEEPALIVEs. +async fn expect_notification(peer: &mut TestPeer) -> Result { + tokio::time::timeout(TIMEOUT, async { + loop { + match peer.recv().await? { + Message::Notification(n) => return Ok(n), + Message::KeepAlive => {} + other => eyre::bail!("expected NOTIFICATION, got {other:?}"), + } + } + }) + .await? +} + +fn is_collision_cease(n: &NotificationMessage) -> bool { + matches!(n.error, BgpError::Cease(CeaseSubcode::ConnCollisionResolution)) +} + +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(()) +} + +/// Peer has the higher identifier, so the connection the *peer* initiated +/// must survive: the server closes its own outbound connection. +#[tokio::test] +async fn test_simultaneous_open_peer_higher_keeps_inbound() -> Result<()> { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let server = server_dialing(&listener).await?; + + // Server's outbound connection reaches OpenConfirm. + let (stream, _) = tokio::time::timeout(TIMEOUT, listener.accept()).await??; + let mut dialled = TestPeer::from_stream(stream); + open_only(&mut dialled, HIGH_ID).await?; + + // The peer now also dials the server. + let mut inbound = TestPeer::connect(server.addr).await?; + open_only(&mut inbound, HIGH_ID).await?; + + let notif = expect_notification(&mut dialled).await?; + assert!(is_collision_cease(¬if), "{notif:?}"); + assert!(dialled.is_closed().await); + + confirm(&mut inbound).await?; + wait_for_sessions(&server, 1).await?; + let session = server.registry.list().remove(0); + assert_eq!(session.direction, Direction::Inbound); + assert_eq!(session.name.as_deref(), Some("lab")); + + // The survivor is fully functional. + server + .local_rib + .announce(LocalRoute::new(IpPrefix::try_from("10.100.1.0/24")?, "alice", LocalRouteAttrs::default())) + .await?; + let update = tokio::time::timeout(TIMEOUT, inbound.recv_update()).await??; + assert_eq!(update.announced_routes[0].to_string(), "10.100.1.0/24"); + + // The dialer notices the surviving session and does not open another. + let extra = tokio::time::timeout(Duration::from_millis(700), listener.accept()).await; + assert!(extra.is_err(), "dialer opened a second connection despite an established session"); + Ok(()) +} + +/// Peer has the lower identifier, so the connection the *server* initiated +/// must survive: the peer's inbound connection is refused. +#[tokio::test] +async fn test_simultaneous_open_peer_lower_keeps_outbound() -> Result<()> { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let server = server_dialing(&listener).await?; + + let (stream, _) = tokio::time::timeout(TIMEOUT, listener.accept()).await??; + let mut dialled = TestPeer::from_stream(stream); + open_only(&mut dialled, LOW_ID).await?; + + let mut inbound = TestPeer::connect(server.addr).await?; + open_only(&mut inbound, LOW_ID).await?; + + let notif = expect_notification(&mut inbound).await?; + assert!(is_collision_cease(¬if), "{notif:?}"); + assert!(inbound.is_closed().await); + + confirm(&mut dialled).await?; + wait_for_sessions(&server, 1).await?; + assert_eq!(server.registry.list()[0].direction, Direction::Outbound); + Ok(()) +} + +/// The order in which the two OPENs are processed must not change the +/// outcome: here the inbound connection reaches OpenConfirm first. +#[tokio::test] +async fn test_outcome_is_independent_of_arrival_order() -> Result<()> { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let server = server_dialing(&listener).await?; + + let (stream, _) = tokio::time::timeout(TIMEOUT, listener.accept()).await??; + let mut dialled = TestPeer::from_stream(stream); + + // Inbound first this time; peer is lower so outbound must still win. + let mut inbound = TestPeer::connect(server.addr).await?; + open_only(&mut inbound, LOW_ID).await?; + open_only(&mut dialled, LOW_ID).await?; + + let notif = expect_notification(&mut inbound).await?; + assert!(is_collision_cease(¬if), "{notif:?}"); + + confirm(&mut dialled).await?; + wait_for_sessions(&server, 1).await?; + assert_eq!(server.registry.list()[0].direction, Direction::Outbound); + Ok(()) +} + +/// RFC default: a collision with an Established session closes the new +/// connection, whatever the identifiers say. +#[tokio::test] +async fn test_established_session_wins_by_default() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + + let mut first = TestPeer::connect(server.addr).await?; + first.handshake(PEER_ASN, HIGH_ID, 90).await?; + wait_for_sessions(&server, 1).await?; + + let mut second = TestPeer::connect(server.addr).await?; + open_only(&mut second, HIGH_ID).await?; + let notif = expect_notification(&mut second).await?; + assert!(is_collision_cease(¬if), "{notif:?}"); + + // The original is untouched. + assert_eq!(server.registry.list().len(), 1); + server + .local_rib + .announce(LocalRoute::new(IpPrefix::try_from("10.100.1.0/24")?, "alice", LocalRouteAttrs::default())) + .await?; + let update = tokio::time::timeout(TIMEOUT, first.recv_update()).await??; + assert_eq!(update.announced_routes[0].to_string(), "10.100.1.0/24"); + Ok(()) +} + +/// With collision_detect_established_state, a peer that restarts and dials +/// again replaces the stale session at once, and the routes learned over the +/// old session are flushed. +#[tokio::test] +async fn test_restarted_peer_replaces_stale_session_when_enabled() -> Result<()> { + let server = TestServer::start_with(LOCAL_ASN, |c| c.collision_detect_established_state = true).await?; + + let mut stale = TestPeer::connect(server.addr).await?; + stale.handshake(PEER_ASN, HIGH_ID, 90).await?; + wait_for_sessions(&server, 1).await?; + stale + .send(&Message::Update(UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![ + PathAttribute::Origin(OriginPathAttribute::IGP), + AsPathAttribute::from_asns(vec![PEER_ASN]), + PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(10, 0, 0, 9))), + ], + announced_routes: vec![IpPrefix::try_from("203.0.113.0/24")?], + })) + .await?; + tokio::time::timeout(TIMEOUT, async { + while server.adj_rib_in.snapshot().await.is_empty() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await?; + let stale_addr = server.registry.list()[0].peer.addr; + + // "Restart": a fresh connection from the same router. + let mut fresh = TestPeer::connect(server.addr).await?; + fresh.handshake(PEER_ASN, HIGH_ID, 90).await?; + + let notif = expect_notification(&mut stale).await?; + assert!(is_collision_cease(¬if), "{notif:?}"); + + tokio::time::timeout(TIMEOUT, async { + loop { + let sessions = server.registry.list(); + if sessions.len() == 1 && sessions[0].peer.addr != stale_addr { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await?; + assert!(server.adj_rib_in.snapshot().await.is_empty(), "stale routes were not flushed"); + Ok(()) +} + +/// Different routers (different identifiers) never collide. +#[tokio::test] +async fn test_distinct_identifiers_do_not_collide() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + let mut a = TestPeer::connect(server.addr).await?; + a.handshake(PEER_ASN, HIGH_ID, 90).await?; + let mut b = TestPeer::connect(server.addr).await?; + b.handshake(PEER_ASN, LOW_ID, 90).await?; + wait_for_sessions(&server, 2).await?; + Ok(()) +} + +/// A failed connection frees its slot: the same router can come back. +#[tokio::test] +async fn test_identifier_is_released_when_connection_ends() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + let mut first = TestPeer::connect(server.addr).await?; + first.handshake(PEER_ASN, HIGH_ID, 90).await?; + wait_for_sessions(&server, 1).await?; + drop(first); + wait_for_sessions(&server, 0).await?; + + let mut second = TestPeer::connect(server.addr).await?; + second.handshake(PEER_ASN, HIGH_ID, 90).await?; + wait_for_sessions(&server, 1).await?; + Ok(()) +} + +/// A peer claiming our own identifier is refused outright. +#[tokio::test] +async fn test_peer_with_our_identifier_is_refused() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + let mut peer = TestPeer::connect(server.addr).await?; + open_only(&mut peer, Ipv4Addr::new(127, 0, 0, 1)).await?; + let notif = expect_notification(&mut peer).await?; + assert!(matches!(notif.error, BgpError::OpenMsg(OpenMsgSubcode::BadBgpId)), "{notif:?}"); + Ok(()) +} diff --git a/crates/server/proto/config.proto b/crates/server/proto/config.proto index 367a063..c027efa 100644 --- a/crates/server/proto/config.proto +++ b/crates/server/proto/config.proto @@ -28,6 +28,15 @@ message BgpConfig { // Initial delay between attempts to connect to a configured peer, in // seconds. Doubles after each failure up to two minutes. Defaults to 5. uint32 connect_retry_seconds = 7; + + // Connection collisions (RFC 4271 §6.8) against a session that is already + // Established are, by default, resolved by closing the new connection. + // When true they are resolved like any other collision: the connection + // initiated by the higher BGP Identifier survives, and a second connection + // opened by the same side replaces the first. This lets a peer that + // restarted take over from a session we have not yet noticed is dead, + // instead of being refused until the hold timer expires. + bool collision_detect_established_state = 8; } // Limits on locally originated routes. diff --git a/crates/server/src/collision.rs b/crates/server/src/collision.rs new file mode 100644 index 0000000..54b197b --- /dev/null +++ b/crates/server/src/collision.rs @@ -0,0 +1,215 @@ +//! Connection collision detection (RFC 4271 §6.8). +//! +//! Two speakers that dial each other can end up with two TCP connections. +//! Exactly one must survive and both ends must pick the same one: the +//! connection *initiated by the speaker with the higher BGP Identifier*. +//! The other is closed with Cease / Connection Collision Resolution. +//! +//! This cannot be a check local to one connection: the rival connection is +//! usually still mid-handshake (so it is not in the session registry yet), +//! and sometimes it is the rival, running in another task, that must close. +//! Hence one small shared map, keyed by the peer's BGP Identifier, holding +//! the single connection currently preferred for that peer and a way to +//! tell it to stop. +//! +//! A collision with an *Established* connection closes the new one, unless +//! `detect_established` is set (the RFC's optional +//! `CollisionDetectEstablishedState`), which lets a restarted peer replace a +//! session we have not yet noticed is dead. If both connections were opened +//! by the same side the newer wins: the initiator has abandoned the older. + +use std::collections::HashMap; +use std::net::Ipv4Addr; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; + +use tokio::sync::oneshot; + +use crate::metrics::COLLISIONS; + +/// Which side opened the TCP connection. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Direction { + /// The peer connected to us. + Inbound, + /// We connected to the peer. + Outbound, +} + +impl Direction { + pub fn as_str(&self) -> &'static str { + match self { + Direction::Inbound => "inbound", + Direction::Outbound => "outbound", + } + } +} + +struct Entry { + id: u64, + direction: Direction, + established: bool, + kill: Option>, +} + +/// True if the new connection is the one to close. +fn new_loses( + local_id: Ipv4Addr, + remote_id: Ipv4Addr, + existing: &Entry, + new: Direction, + detect_established: bool, +) -> bool { + if existing.established && !detect_established { + return true; + } + if existing.direction == new { + return false; + } + // Identifiers compare as unsigned 32-bit integers. + let keep = if u32::from(local_id) > u32::from(remote_id) { Direction::Outbound } else { Direction::Inbound }; + new != keep +} + +/// Shared by every connection of one server. +#[derive(Clone)] +pub struct CollisionTable { + entries: Arc>>, + next_id: Arc, + detect_established: bool, +} + +impl CollisionTable { + pub fn new(detect_established: bool) -> Self { + Self { entries: Default::default(), next_id: Default::default(), detect_established } + } + + /// Call when a connection receives the peer's OPEN. `None` means this + /// connection lost and must be closed with Cease / Connection Collision + /// Resolution. Otherwise any rival has been told to close and this + /// connection holds the slot until the guard drops. + pub fn admit(&self, local_id: Ipv4Addr, remote_id: Ipv4Addr, direction: Direction) -> Option { + let mut entries = self.entries.lock().expect("collision table poisoned"); + if let Some(existing) = entries.get_mut(&remote_id) { + if new_loses(local_id, remote_id, existing, direction, self.detect_established) { + COLLISIONS.with_label_values(&["new"]).inc(); + return None; + } + if let Some(kill) = existing.kill.take() { + COLLISIONS.with_label_values(&["existing"]).inc(); + let _ = kill.send(()); + } + } + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + let (kill, killed) = oneshot::channel(); + entries.insert(remote_id, Entry { id, direction, established: false, kill: Some(kill) }); + Some(ConnectionGuard { table: self.clone(), remote_id, id, killed }) + } +} + +/// Holds a connection's slot in the table; releases it on drop. +pub struct ConnectionGuard { + table: CollisionTable, + remote_id: Ipv4Addr, + id: u64, + /// Completes when a later connection won against this one; the owner + /// must then send Cease / Connection Collision Resolution and close. + pub killed: oneshot::Receiver<()>, +} + +impl ConnectionGuard { + pub fn set_established(&self) { + let mut entries = self.table.entries.lock().expect("collision table poisoned"); + if let Some(e) = entries.get_mut(&self.remote_id).filter(|e| e.id == self.id) { + e.established = true; + } + } +} + +impl Drop for ConnectionGuard { + fn drop(&mut self) { + // A connection that lost has already been replaced in the map; only + // remove the slot if it is still ours. + let mut entries = self.table.entries.lock().expect("collision table poisoned"); + if entries.get(&self.remote_id).is_some_and(|e| e.id == self.id) { + entries.remove(&self.remote_id); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use Direction::*; + + const LOW: Ipv4Addr = Ipv4Addr::new(10, 0, 0, 1); + const HIGH: Ipv4Addr = Ipv4Addr::new(192, 0, 2, 1); + + fn loses(local: Ipv4Addr, remote: Ipv4Addr, existing: Direction, established: bool, new: Direction, detect: bool) -> bool { + let e = Entry { id: 0, direction: existing, established, kill: None }; + new_loses(local, remote, &e, new, detect) + } + + #[test] + fn higher_identifier_keeps_the_connection_it_initiated() { + // We are HIGH: our outbound connection survives. + assert!(loses(HIGH, LOW, Outbound, false, Inbound, false)); + assert!(!loses(HIGH, LOW, Inbound, false, Outbound, false)); + // We are LOW: the peer's connection (inbound to us) survives. + assert!(!loses(LOW, HIGH, Outbound, false, Inbound, false)); + assert!(loses(LOW, HIGH, Inbound, false, Outbound, false)); + } + + #[test] + fn both_ends_agree() { + // A and B dial each other. Each end holds its own dial as the + // existing (Outbound) connection and sees the other's dial arrive as + // new (Inbound). Exactly one of the two dials may be closed, so the + // ends must give opposite answers to "close the other side's dial?". + for (a, b) in [(LOW, HIGH), (HIGH, LOW)] { + let a_closes_dial_from_b = loses(a, b, Outbound, false, Inbound, false); + let b_closes_dial_from_a = loses(b, a, Outbound, false, Inbound, false); + assert_ne!(a_closes_dial_from_b, b_closes_dial_from_a, "ends disagree for a={a} b={b}"); + } + } + + #[test] + fn established_wins_unless_enabled() { + for new in [Inbound, Outbound] { + for existing in [Inbound, Outbound] { + assert!(loses(LOW, HIGH, existing, true, new, false)); + assert!(loses(HIGH, LOW, existing, true, new, false)); + } + } + assert!(!loses(LOW, HIGH, Outbound, true, Inbound, true)); + assert!(loses(HIGH, LOW, Outbound, true, Inbound, true)); + } + + #[test] + fn same_direction_newer_wins() { + assert!(!loses(LOW, HIGH, Inbound, false, Inbound, false)); + assert!(!loses(HIGH, LOW, Inbound, false, Inbound, false)); + assert!(loses(LOW, HIGH, Inbound, true, Inbound, false)); + assert!(!loses(LOW, HIGH, Inbound, true, Inbound, true)); + } + + #[tokio::test] + async fn table_signals_loser_and_releases_slots() { + let table = CollisionTable::new(false); + // We are LOW: outbound first, then the peer's inbound wins. + let mut out = table.admit(LOW, HIGH, Outbound).unwrap(); + let inb = table.admit(LOW, HIGH, Inbound).unwrap(); + assert!(out.killed.try_recv().is_ok()); + // The loser's guard dropping must not free the winner's slot. + drop(out); + inb.set_established(); + assert!(table.admit(LOW, HIGH, Outbound).is_none()); + + // A different identifier never collides. + assert!(table.admit(LOW, Ipv4Addr::new(192, 0, 2, 2), Inbound).is_some()); + + // Once the survivor goes away the identifier is free again. + drop(inb); + assert!(table.admit(LOW, HIGH, Outbound).is_some()); + } +} diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index a097922..9ba9e3b 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -23,6 +23,10 @@ pub struct Config { /// Initial delay between attempts to connect to a configured peer; /// doubles on each failure. pub connect_retry: Duration, + /// Resolve collisions against Established sessions by BGP Identifier + /// instead of always closing the new connection (RFC 4271 §8.1.1 + /// `CollisionDetectEstablishedState`). + pub collision_detect_established_state: bool, /// Limits on locally originated routes. pub local_route_policy: LocalRibPolicy, /// Address for the HTTP API; `None` disables it. @@ -121,6 +125,7 @@ impl Config { listen_endpoints, peers, connect_retry, + collision_detect_established_state: proto.collision_detect_established_state, local_route_policy, http_endpoint, }) diff --git a/crates/server/src/dialer.rs b/crates/server/src/dialer.rs index 6ace5d8..b15fe14 100644 --- a/crates/server/src/dialer.rs +++ b/crates/server/src/dialer.rs @@ -3,9 +3,10 @@ //! One task per configured peer loops forever: connect, run the session, //! back off, repeat. While a session with the same remote IP and ASN is //! already established (for example because the peer dialled us first) the -//! dialer stays idle rather than opening a second one. Full RFC 4271 §6.8 -//! collision detection is not implemented; a real router on the other end -//! will resolve any remaining race by closing one of the two connections. +//! dialer stays idle rather than opening a second one. When both sides dial +//! at once the FSM resolves the collision (see [`crate::collision`]); if our +//! connection is the one closed, the next pass finds the surviving session +//! and goes idle. use std::sync::Arc; use std::time::{Duration, SystemTime}; @@ -14,6 +15,7 @@ use tokio::net::TcpStream; use tokio::task::JoinHandle; use tracing::{info, warn}; +use crate::collision::Direction; use crate::config::PeerConfig; use crate::fsm::{SessionConfig, run_session}; use crate::session::{DialState, SessionRegistry}; @@ -32,6 +34,7 @@ pub fn spawn_dialers( .map(|peer| { let config = SessionConfig { hold_time_secs: peer.hold_time_secs, + direction: Direction::Outbound, peer_name: Some(peer.name.clone()), expected_asn: Some(peer.remote_asn), ..base.clone() diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index a56964e..f0debbe 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -1,7 +1,7 @@ use std::net::{Ipv4Addr, SocketAddr}; use std::time::Duration; -use bgp_packet::errors::{BgpError, FsmSubcode, OpenMsgSubcode}; +use bgp_packet::errors::{BgpError, CeaseSubcode, 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}; @@ -12,12 +12,13 @@ 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::sync::{mpsc, oneshot}; use tokio::time::{Instant, interval_at, sleep_until}; use tracing::{info, warn}; use bgp_packet::message::{AS_TRANS, BGP4_VERSION}; +use crate::collision::{CollisionTable, Direction}; use crate::metrics::{MESSAGES_RECEIVED, MESSAGES_SENT, SESSIONS_ESTABLISHED_TOTAL, message_type_name}; use crate::session::{SessionHandle, SessionRegistry}; @@ -33,6 +34,10 @@ pub struct SessionConfig { pub hold_time_secs: u16, pub adj_rib_in: AdjRibIn, pub registry: SessionRegistry, + /// Shared table used to resolve connection collisions (RFC 4271 §6.8). + pub collisions: CollisionTable, + /// Which side opened this TCP connection. + pub direction: Direction, /// Configured name of the peer, if known. pub peer_name: Option, /// If set, the peer's OPEN must carry this ASN or the session is @@ -87,7 +92,7 @@ pub async fn run_session( } }; - if let Err(bgp_err) = validate_open(&peer_open) { + if let Err(bgp_err) = validate_open(&peer_open, config.router_id) { send_notification(&mut writer, bgp_err, &ParserContext::default()).await; bail!("Peer {} sent invalid OPEN", peer_addr); } @@ -124,19 +129,39 @@ pub async fn run_session( bail!("Peer {peer_addr} announced AS{peer_asn} but AS{expected} was configured"); } + // Connection collision detection (RFC 4271 §6.8). From here on the + // guard keeps this connection visible to later ones, and `killed` fires + // if one of them wins. + let Some(mut collision_guard) = + config.collisions.admit(config.router_id, peer_open.identifier, config.direction) + else { + return close_for_collision(&mut writer, &ctx, peer_addr).await; + }; + info!( %peer_addr, peer_asn, negotiated_hold_time, + direction = config.direction.as_str(), "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); + // With a negotiated hold time of zero the hold timer does not run + // (RFC 4271 §4.2); fall back to the large pre-negotiation value so a + // silent peer is still cleaned up eventually. + let hold_duration = Duration::from_secs(if negotiated_hold_time == 0 { 240 } else { negotiated_hold_time as u64 }); - match recv_with_timeout(&mut reader, &ctx, hold_duration).await { + let confirm = tokio::select! { + msg = recv_with_timeout(&mut reader, &ctx, hold_duration) => msg, + _ = &mut collision_guard.killed => { + return close_for_collision(&mut writer, &ctx, peer_addr).await; + } + }; + + match confirm { Ok(Message::KeepAlive) => {} Ok(other) => { send_notification( @@ -162,9 +187,11 @@ pub async fn run_session( // 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); + collision_guard.set_established(); let _guard = config.registry.register(SessionHandle::new( peer_id, config.peer_name.clone(), + config.direction, local_addr, peer_four_byte_asn, outbound_tx, @@ -177,7 +204,7 @@ pub async fn run_session( negotiated_hold_time, peer_id, &config.adj_rib_in, - &mut outbound_rx, + SessionInputs { outbound: &mut outbound_rx, killed: &mut collision_guard.killed }, ) .await; @@ -189,6 +216,14 @@ pub async fn run_session( result } +/// Signals from the rest of the server into an established session. +struct SessionInputs<'a> { + /// Messages queued for transmission to the peer. + outbound: &'a mut mpsc::Receiver, + /// Fires when a later connection won collision resolution. + killed: &'a mut oneshot::Receiver<()>, +} + async fn established_loop( reader: &mut OwnedReadHalf, writer: &mut OwnedWriteHalf, @@ -196,8 +231,9 @@ async fn established_loop( hold_time_secs: u16, peer_id: PeerId, adj_rib_in: &AdjRibIn, - outbound: &mut mpsc::Receiver, + inputs: SessionInputs<'_>, ) -> Result<()> { + let SessionInputs { outbound, killed } = inputs; // 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. @@ -255,6 +291,9 @@ async fn established_loop( } send_message(writer, &msg, ctx).await?; } + _ = &mut *killed => { + return close_for_collision(writer, ctx, peer_id.addr).await; + } _ = keepalive_timer.tick(), if timers_enabled => { send_message(writer, &Message::KeepAlive, ctx).await?; } @@ -326,6 +365,18 @@ async fn send_notification(writer: &mut OwnedWriteHalf, error: BgpError, ctx: &P } } +/// A later connection to the same BGP Identifier won collision resolution: +/// tell the peer why and end this session. +async fn close_for_collision( + writer: &mut OwnedWriteHalf, + ctx: &ParserContext, + peer_addr: SocketAddr, +) -> Result<()> { + info!(%peer_addr, "Closing connection: lost collision resolution"); + send_notification(writer, BgpError::Cease(CeaseSubcode::ConnCollisionResolution), ctx).await; + bail!("Connection to {peer_addr} closed by collision resolution"); +} + // --------------------------------------------------------------------------- // RIB helpers // --------------------------------------------------------------------------- @@ -398,10 +449,15 @@ fn build_open(config: &SessionConfig) -> Message { }) } -fn validate_open(open: &OpenMessage) -> Result<(), BgpError> { +fn validate_open(open: &OpenMessage, local_id: Ipv4Addr) -> Result<(), BgpError> { if open.version != BGP4_VERSION { return Err(BgpError::OpenMsg(OpenMsgSubcode::UnsupportedVersion)); } + // A peer cannot share our identifier or use the unspecified address; + // collision resolution depends on identifiers being distinct. + if open.identifier == local_id || open.identifier.is_unspecified() { + return Err(BgpError::OpenMsg(OpenMsgSubcode::BadBgpId)); + } if open.hold_time != 0 && open.hold_time < 3 { return Err(BgpError::OpenMsg(OpenMsgSubcode::UnacceptableHoldTime)); } diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 9af137f..3dd9ee0 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -1,4 +1,5 @@ pub mod advertise; +pub mod collision; pub mod config; pub mod connection; pub mod dialer; diff --git a/crates/server/src/metrics.rs b/crates/server/src/metrics.rs index 11d0175..37ce935 100644 --- a/crates/server/src/metrics.rs +++ b/crates/server/src/metrics.rs @@ -28,6 +28,15 @@ pub static SESSIONS_ESTABLISHED: LazyLock = LazyLock::new(|| { .expect("metric registration") }); +pub static COLLISIONS: LazyLock = LazyLock::new(|| { + register_int_counter_vec!( + "bgp_connection_collisions_total", + "Connection collisions resolved, by which connection was closed", + &["closed"] + ) + .expect("metric registration") +}); + pub static LOCAL_ROUTES: LazyLock = LazyLock::new(|| { register_int_gauge!("bgp_local_routes", "Locally originated routes").expect("metric registration") }); @@ -44,6 +53,7 @@ pub fn init() { LazyLock::force(&MESSAGES_RECEIVED); LazyLock::force(&SESSIONS_ESTABLISHED_TOTAL); LazyLock::force(&SESSIONS_ESTABLISHED); + LazyLock::force(&COLLISIONS); LazyLock::force(&LOCAL_ROUTES); LazyLock::force(&ADJ_RIB_IN_ROUTES); } diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index 3032658..306b676 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -10,6 +10,7 @@ use tracing::{error, info}; use bgp_rib::{AdjRibIn, LocalRib}; use crate::advertise::spawn_advertiser; +use crate::collision::{CollisionTable, Direction}; use crate::config::Config; use crate::connection::handle_connection; use crate::dialer::spawn_dialers; @@ -98,6 +99,8 @@ impl BgpServer { hold_time_secs: self.config.hold_time_secs, adj_rib_in: self.adj_rib_in.clone(), registry: self.registry.clone(), + collisions: CollisionTable::new(self.config.collision_detect_established_state), + direction: Direction::Inbound, peer_name: None, expected_asn: None, }; diff --git a/crates/server/src/session.rs b/crates/server/src/session.rs index aa72d8c..5162362 100644 --- a/crates/server/src/session.rs +++ b/crates/server/src/session.rs @@ -14,6 +14,7 @@ use bgp_packet::message::Message; use bgp_rib::PeerId; use tokio::sync::{broadcast, mpsc}; +use crate::collision::Direction; use crate::metrics::SESSIONS_ESTABLISHED; const EVENT_CHANNEL_CAPACITY: usize = 256; @@ -42,6 +43,8 @@ pub struct SessionHandle { pub peer: PeerId, /// Name from the configuration, if this peer is configured. pub name: Option, + /// Which side opened the TCP connection. + pub direction: Direction, /// Our side of the TCP connection; used as the next hop for routes we /// originate to this peer. pub local_addr: SocketAddr, @@ -60,11 +63,12 @@ impl SessionHandle { pub fn new( peer: PeerId, name: Option, + direction: Direction, local_addr: SocketAddr, four_octet_asn: bool, outbound: mpsc::Sender, ) -> Self { - Self { peer, name, local_addr, established_at: SystemTime::now(), four_octet_asn, outbound } + Self { peer, name, direction, local_addr, established_at: SystemTime::now(), four_octet_asn, outbound } } /// Queues a message for transmission on this session. Waits for queue diff --git a/crates/web/src/api.rs b/crates/web/src/api.rs index a2c520a..e353730 100644 --- a/crates/web/src/api.rs +++ b/crates/web/src/api.rs @@ -83,6 +83,8 @@ pub struct PeerView { pub peer_asn: u32, /// `Established`, `Connecting`, `Backoff` or `Idle`. pub state: String, + /// `inbound` or `outbound`: which side opened the live TCP connection. + pub direction: Option, pub local_addr: Option, pub established_at_unix: Option, pub uptime_secs: Option, @@ -99,6 +101,7 @@ impl From<&SessionHandle> for PeerView { peer_addr: s.peer.addr.to_string(), peer_asn: s.peer.asn, state: "Established".into(), + direction: Some(s.direction.as_str().into()), local_addr: Some(s.local_addr.to_string()), established_at_unix: Some(unix_secs(s.established_at)), uptime_secs: Some(s.established_at.elapsed().map(|d| d.as_secs()).unwrap_or(0)), @@ -121,6 +124,7 @@ impl PeerView { peer_addr: peer.endpoint.to_string(), peer_asn: peer.remote_asn, state: state_name.into(), + direction: None, local_addr: None, established_at_unix: None, uptime_secs: None, diff --git a/crates/web/static/statusz.html b/crates/web/static/statusz.html index ba9534d..e98be1d 100644 --- a/crates/web/static/statusz.html +++ b/crates/web/static/statusz.html @@ -82,7 +82,7 @@

Peers

- +
namepeerASNstatelocal address (next hop)establisheduptime4-byte ASNdetail
namepeerASNstateopened bylocal address (next hop)establisheduptime4-byte ASNdetail
@@ -170,13 +170,13 @@ const STATE_CLASS = { Established: 'ok', Connecting: 'warn', Backoff: 'err', Idle: 'muted' }; function renderPeers() { const b = $('peers-body'); $('peers-count').textContent = `${state.peers.filter(p => p.state === 'Established').length} up / ${state.peers.length}`; - if (!state.peers.length) { b.innerHTML = 'No peers configured and none connected.'; return; } + if (!state.peers.length) { b.innerHTML = 'No peers configured and none connected.'; return; } b.innerHTML = state.peers.map(p => { const up = p.state === 'Established'; const detail = p.state === 'Backoff' ? `${esc(p.last_error || '')}; retry in ${fmtDur(p.retry_at_unix - now())}` : ''; return ` ${esc(p.name ?? '')}${p.name ? '' : '(inbound)'}${esc(p.peer_addr)}${p.peer_asn} - ${esc(p.state)}${up ? esc(p.local_addr) : '—'} + ${esc(p.state)}${up ? (p.direction === 'outbound' ? 'us' : 'peer') : '—'}${up ? esc(p.local_addr) : '—'} ${up ? fmtTime(p.established_at_unix) : '—'}${up ? age(p.established_at_unix) : '—'}${up ? (p.four_octet_asn ? 'yes' : 'no') : '—'} ${detail}`; }).join(''); } diff --git a/docs/RFCs.md b/docs/RFCs.md index c65bcca..a1857b7 100644 --- a/docs/RFCs.md +++ b/docs/RFCs.md @@ -27,7 +27,9 @@ The base protocol specification. | §8 | FSM — passive (inbound) side: OpenSent → OpenConfirm → Established | ✅ | | §8 | FSM — active (outbound) side: dial configured peers, exponential retry | ✅ | | §6.2 | Bad Peer AS check against configured ASN | ✅ | -| §6.8 | Connection collision detection (router-ID comparison) | ❌ | +| §6.8 | Connection collision detection (BGP Identifier comparison, Cease / Connection Collision Resolution) | ✅ | +| §8.1.1 | Optional `CollisionDetectEstablishedState` (`collision_detect_established_state`) | ✅ | +| §6.2 | Bad BGP Identifier check (peer using our identifier, or 0.0.0.0) | ✅ | | §9 | Decision process / best-path selection | ❌ | | §9 | Adj-RIB-In (per-peer table of received routes) | ✅ | | §9 | Locally originated routes, advertised to all established peers | ✅ | diff --git a/docs/lab/README.md b/docs/lab/README.md index 0a834a1..770af42 100644 --- a/docs/lab/README.md +++ b/docs/lab/README.md @@ -72,5 +72,11 @@ they have not been executed. Things most likely to need a touch: capability there. If you run it on the host instead, either run as root, grant `CAP_NET_BIND_SERVICE`, or change `listen_endpoints` and the FRR neighbor port. +- Both ferrix and the FRR routers dial each other, so connection collisions + (RFC 4271 §6.8) are normal at start-up: one of the two TCP connections is + closed with a Cease / Connection Collision Resolution NOTIFICATION and the + other carries the session. ferrix has the lowest BGP Identifier here, so + the connection the FRR router opened is the one that survives; the + "opened by" column on the status page shows which side that was. - ferrix does not check AS paths for its own ASN. That is what lets the looped route be accepted; a production BGP speaker would reject it. diff --git a/docs/lab/ferrix.textpb b/docs/lab/ferrix.textpb index 95bada8..755e9af 100644 --- a/docs/lab/ferrix.textpb +++ b/docs/lab/ferrix.textpb @@ -16,6 +16,11 @@ peers { } connect_retry_seconds: 5 +# If r1 or r2 is restarted it will dial us again while we still believe the +# old session is up. Let that new connection replace the stale one straight +# away instead of refusing it until the hold timer (30 s here) expires. +collision_detect_established_state: true + # Students may only announce out of 10.100.0.0/16; one /24 each is plenty. local_route_policy { allowed_prefixes: "10.100.0.0/16" diff --git a/testdata/server.textpb b/testdata/server.textpb index c9b81dc..5b86f71 100644 --- a/testdata/server.textpb +++ b/testdata/server.textpb @@ -15,6 +15,9 @@ peers { hold_time_seconds: 30 } connect_retry_seconds: 5 +# Let a restarted peer replace a stale Established session immediately +# (RFC 4271 CollisionDetectEstablishedState). Default false. +collision_detect_established_state: false local_route_policy { allowed_prefixes: "10.100.0.0/16" allowed_prefixes: "2001:db8:100::/48"