diff --git a/README.md b/README.md index 46968ef..894afd7 100644 --- a/README.md +++ b/README.md @@ -20,8 +20,10 @@ We aim for the 1.0 release to be functional and stable to operate peering sessio - ✅ Capabilities from RFC4271 - ✅ Update message - ☐ Notification message -- ☐ Implement scaffolding required to start and configure a server -- ☐ Implement connection logic to establish sessions with peers +- ✅ Implement scaffolding required to start and configure a server +- ✅ Implement connection logic to establish sessions with peers (passive and active) +- ✅ Locally originated routes with communities, announced to all peers +- ✅ HTTP API, live status page and Prometheus metrics - ☐ Implement an efficient data structure for storing prefixes (e.g. Poptrie) - ☐ Filtering for peer RIB in / out - ☐ Main / auxillary RIBs for exchanging routes between peers diff --git a/crates/integration-tests/examples/fake_peer.rs b/crates/integration-tests/examples/fake_peer.rs index cb42d6f..1ce2de7 100644 --- a/crates/integration-tests/examples/fake_peer.rs +++ b/crates/integration-tests/examples/fake_peer.rs @@ -8,6 +8,9 @@ //! With `--echo`, every prefix received is re-announced back to the server //! with this peer's ASN prepended, which lets a single ferrix instance show //! its own routes "coming back" in the received table. +//! +//! With `--listen ADDR` it instead waits for the server to dial it (one +//! connection at a time), which exercises the server's active side. use std::net::Ipv4Addr; use std::time::Duration; @@ -20,12 +23,17 @@ use bgp_packet::message::{ use bgp_primitives::ip_prefix::IpPrefix; use clap::Parser; use eyre::Result; +use tokio::net::TcpListener; #[derive(Parser, Debug)] struct Args { - /// Address of the ferrix_bgp listener. - #[arg(long, default_value = "127.0.0.1:1790")] + /// Address of the ferrix_bgp listener to connect to. + #[arg(long, default_value = "127.0.0.1:1790", conflicts_with = "listen")] server: String, + /// Instead of connecting, listen here and accept connections from the + /// server, one at a time, forever. + #[arg(long)] + listen: Option, /// ASN this peer speaks as. #[arg(long, default_value_t = 65001)] asn: u32, @@ -49,8 +57,26 @@ const NO_EXPORT: (u16, u16) = (0xFFFF, 0xFF01); #[tokio::main] async fn main() -> Result<()> { let args = Args::parse(); - let addr = args.server.parse()?; - let mut peer = TestPeer::connect(addr).await?; + match &args.listen { + Some(listen) => { + let listener = TcpListener::bind(listen).await?; + println!("listening on {}", listener.local_addr()?); + loop { + let (stream, from) = listener.accept().await?; + println!("accepted connection from {from}"); + if let Err(e) = run(&args, TestPeer::from_stream(stream), from).await { + println!("session ended: {e}"); + } + } + } + None => { + let addr = args.server.parse()?; + run(&args, TestPeer::connect(addr).await?, addr).await + } + } +} + +async fn run(args: &Args, mut peer: TestPeer, addr: std::net::SocketAddr) -> Result<()> { let open = peer.handshake(args.asn, args.router_id, args.hold_time).await?; println!("established with {} (AS{} router-id {})", addr, open.asn, open.identifier); @@ -80,7 +106,7 @@ async fn main() -> Result<()> { } } Message::Notification(n) => { - println!("NOTIFICATION {}; exiting", n.error); + println!("NOTIFICATION {}; session over", n.error); return Ok(()); } other => println!("unexpected {other:?}"), diff --git a/crates/integration-tests/src/test_peer.rs b/crates/integration-tests/src/test_peer.rs index 380763b..ad19ed6 100644 --- a/crates/integration-tests/src/test_peer.rs +++ b/crates/integration-tests/src/test_peer.rs @@ -19,7 +19,13 @@ pub struct TestPeer { impl TestPeer { pub async fn connect(addr: SocketAddr) -> Result { let stream = TcpStream::connect(addr).await?; - Ok(Self { stream, ctx: ParserContext::default() }) + Ok(Self::from_stream(stream)) + } + + /// Wraps an already-connected stream, e.g. one accepted from a server + /// that dialled us. + pub fn from_stream(stream: TcpStream) -> Self { + Self { stream, ctx: ParserContext::default() } } /// Send a BGP message. diff --git a/crates/integration-tests/src/test_server.rs b/crates/integration-tests/src/test_server.rs index 7d00a4d..6181a73 100644 --- a/crates/integration-tests/src/test_server.rs +++ b/crates/integration-tests/src/test_server.rs @@ -41,15 +41,27 @@ impl TestServer { } async fn start_inner(local_asn: u32, hold_time_secs: u16, policy: LocalRibPolicy) -> Result { - let config = Config { + Self::start_with(local_asn, |c| { + c.hold_time_secs = hold_time_secs; + c.local_route_policy = policy; + }) + .await + } + + /// Start a server after letting `customize` edit the default test config + /// (e.g. to add configured peers). + pub async fn start_with(local_asn: u32, customize: impl FnOnce(&mut Config)) -> Result { + let mut config = Config { router_id: Ipv4Addr::new(127, 0, 0, 1), local_asn, - hold_time_secs, + hold_time_secs: 90, listen_endpoints: vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)], peers: vec![], - local_route_policy: policy, + connect_retry: std::time::Duration::from_secs(5), + local_route_policy: LocalRibPolicy::default(), http_endpoint: None, }; + customize(&mut config); let bound = BgpServerBuilder::new(config).bind().await?; let addr = bound diff --git a/crates/integration-tests/tests/dialer.rs b/crates/integration-tests/tests/dialer.rs new file mode 100644 index 0000000..2db3289 --- /dev/null +++ b/crates/integration-tests/tests/dialer.rs @@ -0,0 +1,177 @@ +//! The active side: the server dials its configured peers and keeps +//! retrying. + +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, OpenMsgSubcode}; +use bgp_packet::message::Message; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rib::{LocalRoute, LocalRouteAttrs}; +use bgp_server::config::PeerConfig; +use bgp_server::session::DialState; +use eyre::Result; +use tokio::net::TcpListener; + +const LOCAL_ASN: u32 = 65000; +const PEER_ASN: u32 = 65001; +const TIMEOUT: Duration = Duration::from_secs(5); + +/// Starts a server configured to dial `listener`'s address as peer "lab". +async fn server_dialing(listener: &TcpListener, remote_asn: u32) -> 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, hold_time_secs: 90 }]; + }) + .await +} + +async fn accept(listener: &TcpListener) -> Result { + let (stream, _) = tokio::time::timeout(TIMEOUT, listener.accept()).await??; + Ok(TestPeer::from_stream(stream)) +} + +#[tokio::test] +async fn test_server_dials_configured_peer_and_establishes() -> Result<()> { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let server = server_dialing(&listener, PEER_ASN).await?; + + let mut peer = accept(&listener).await?; + // Both sides send OPEN on connect; the test peer's handshake order works + // whichever side opened TCP. + let open = peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90).await?; + assert_eq!(open.asn, LOCAL_ASN as u16); + + let mut events = server.registry.subscribe(); + // Either the Up event is still queued or the session is already listed. + let listed = tokio::time::timeout(TIMEOUT, async { + loop { + if let Some(s) = server.registry.list().first().cloned() { + return s; + } + let _ = events.recv().await; + } + }) + .await?; + assert_eq!(listed.name.as_deref(), Some("lab")); + assert_eq!(listed.peer.asn, PEER_ASN); + assert_eq!(listed.peer.addr, listener.local_addr()?); + + // Configured view reflects it; the dialer is not in backoff. + let (cfg, state) = server.registry.configured_peers().into_iter().next().unwrap(); + assert_eq!(cfg.name, "lab"); + assert!(matches!(state, DialState::Connecting | DialState::Idle), "{state:?}"); + + // Local routes flow over the dialled session too. + server + .local_rib + .announce(LocalRoute::new(IpPrefix::try_from("10.100.1.0/24")?, "alice", LocalRouteAttrs::default())) + .await?; + let update = tokio::time::timeout(TIMEOUT, peer.recv_update()).await??; + assert_eq!(update.announced_routes[0].to_string(), "10.100.1.0/24"); + Ok(()) +} + +#[tokio::test] +async fn test_server_reconnects_after_peer_closes() -> Result<()> { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let server = server_dialing(&listener, PEER_ASN).await?; + + let mut first = accept(&listener).await?; + first.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90).await?; + tokio::time::timeout(TIMEOUT, async { + while server.registry.list().is_empty() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await?; + drop(first); + + // Session goes away, then the dialer comes back. + tokio::time::timeout(TIMEOUT, async { + while !server.registry.list().is_empty() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await?; + let (_, state) = server.registry.configured_peers().into_iter().next().unwrap(); + assert!(matches!(state, DialState::Backoff { .. }), "{state:?}"); + + let mut second = accept(&listener).await?; + second.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90).await?; + tokio::time::timeout(TIMEOUT, async { + while server.registry.list().is_empty() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await?; + Ok(()) +} + +#[tokio::test] +async fn test_wrong_asn_is_refused_with_bad_peer_as() -> Result<()> { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let _server = server_dialing(&listener, PEER_ASN).await?; + + let mut peer = accept(&listener).await?; + peer.send_open(65099, Ipv4Addr::new(10, 0, 0, 2), 90).await?; + let _ = peer.recv_open().await?; + let notif = tokio::time::timeout(TIMEOUT, async { + loop { + match peer.recv().await? { + Message::Notification(n) => return Ok::<_, eyre::Error>(n), + Message::KeepAlive => {} + other => eyre::bail!("unexpected {other:?}"), + } + } + }) + .await??; + assert!(matches!(notif.error, BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs)), "{notif:?}"); + + // The dialer retries rather than giving up. + let _again = accept(&listener).await?; + Ok(()) +} + +/// An inbound connection from a configured peer's address is also checked +/// against the configured ASN and gets the configured name. +#[tokio::test] +async fn test_inbound_from_configured_address_uses_config() -> Result<()> { + // Configure a peer at the loopback address on a port nobody listens on, + // so the dialer never succeeds; then connect inbound from loopback. + let server = TestServer::start_with(LOCAL_ASN, |c| { + c.connect_retry = Duration::from_secs(60); + c.peers = vec![PeerConfig { + name: "lab".into(), + endpoint: SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1), + remote_asn: PEER_ASN, + hold_time_secs: 90, + }]; + }) + .await?; + + // Wrong ASN inbound → refused. + let mut bad = TestPeer::connect(server.addr).await?; + bad.send_open(65099, Ipv4Addr::new(10, 0, 0, 2), 90).await?; + let _ = bad.recv_open().await?; + let msg = tokio::time::timeout(TIMEOUT, bad.recv()).await??; + assert!(matches!(&msg, Message::Notification(n) if matches!(n.error, BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs))), "{msg:?}"); + + // Right ASN inbound → established, named. + let mut good = TestPeer::connect(server.addr).await?; + good.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90).await?; + let listed = tokio::time::timeout(TIMEOUT, async { + loop { + if let Some(s) = server.registry.list().first().cloned() { + return s; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await?; + assert_eq!(listed.name.as_deref(), Some("lab")); + Ok(()) +} diff --git a/crates/server/proto/config.proto b/crates/server/proto/config.proto index b3f419c..367a063 100644 --- a/crates/server/proto/config.proto +++ b/crates/server/proto/config.proto @@ -24,6 +24,10 @@ message BgpConfig { // Where to serve the HTTP API, status page and metrics. The port is // required. Unset disables the HTTP server. TcpEndpoint http_endpoint = 6; + + // 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; } // Limits on locally originated routes. diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index a2fd197..a097922 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -1,5 +1,6 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::str::FromStr; +use std::time::Duration; use bgp_primitives::ip_prefix::IpPrefix; use bgp_rib::LocalRibPolicy; @@ -9,6 +10,7 @@ use crate::proto::config::BgpConfig; const DEFAULT_BGP_PORT: u16 = 179; const DEFAULT_HOLD_TIME_SECS: u16 = 90; +const DEFAULT_CONNECT_RETRY: Duration = Duration::from_secs(5); /// Validated, Rust-native server configuration derived from the proto config. pub struct Config { @@ -18,12 +20,16 @@ pub struct Config { pub hold_time_secs: u16, pub listen_endpoints: Vec, pub peers: Vec, + /// Initial delay between attempts to connect to a configured peer; + /// doubles on each failure. + pub connect_retry: Duration, /// Limits on locally originated routes. pub local_route_policy: LocalRibPolicy, /// Address for the HTTP API; `None` disables it. pub http_endpoint: Option, } +#[derive(Debug, Clone)] pub struct PeerConfig { pub name: String, pub endpoint: SocketAddr, @@ -102,12 +108,19 @@ impl Config { None => None, }; + let connect_retry = if proto.connect_retry_seconds == 0 { + DEFAULT_CONNECT_RETRY + } else { + Duration::from_secs(proto.connect_retry_seconds as u64) + }; + Ok(Self { router_id, local_asn: proto.local_asn, hold_time_secs: DEFAULT_HOLD_TIME_SECS, listen_endpoints, peers, + connect_retry, local_route_policy, http_endpoint, }) diff --git a/crates/server/src/dialer.rs b/crates/server/src/dialer.rs new file mode 100644 index 0000000..6ace5d8 --- /dev/null +++ b/crates/server/src/dialer.rs @@ -0,0 +1,84 @@ +//! Active side of the FSM: dial each configured peer and keep the session up. +//! +//! 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. + +use std::sync::Arc; +use std::time::{Duration, SystemTime}; + +use tokio::net::TcpStream; +use tokio::task::JoinHandle; +use tracing::{info, warn}; + +use crate::config::PeerConfig; +use crate::fsm::{SessionConfig, run_session}; +use crate::session::{DialState, SessionRegistry}; + +/// Upper bound for exponential backoff between connection attempts. +const MAX_BACKOFF: Duration = Duration::from_secs(120); +const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); + +pub fn spawn_dialers( + peers: &[PeerConfig], + base: &SessionConfig, + connect_retry: Duration, +) -> Vec> { + peers + .iter() + .map(|peer| { + let config = SessionConfig { + hold_time_secs: peer.hold_time_secs, + peer_name: Some(peer.name.clone()), + expected_asn: Some(peer.remote_asn), + ..base.clone() + }; + tokio::spawn(dial_loop(Arc::new(peer.clone()), config, connect_retry)) + }) + .collect() +} + +async fn dial_loop(peer: Arc, config: SessionConfig, connect_retry: Duration) { + let registry: SessionRegistry = config.registry.clone(); + let mut backoff = connect_retry; + + loop { + if registry.find(peer.endpoint.ip(), peer.remote_asn).is_some() { + // Peer dialled us; nothing to do until that session goes away. + registry.set_dial_state(&peer.name, DialState::Idle); + tokio::time::sleep(connect_retry).await; + continue; + } + + registry.set_dial_state(&peer.name, DialState::Connecting); + info!(peer = %peer.name, endpoint = %peer.endpoint, "Connecting to peer"); + + let outcome = match tokio::time::timeout(CONNECT_TIMEOUT, TcpStream::connect(peer.endpoint)).await { + Ok(Ok(stream)) => { + info!(peer = %peer.name, endpoint = %peer.endpoint, "TCP connected, starting session"); + backoff = connect_retry; + run_session(stream, peer.endpoint, config.clone()).await.map(|()| "session closed by peer".to_string()) + } + Ok(Err(e)) => Err(eyre::eyre!("connect failed: {e}")), + Err(_) => Err(eyre::eyre!("connect timed out after {CONNECT_TIMEOUT:?}")), + }; + + let reason = match outcome { + Ok(msg) => msg, + Err(e) => { + warn!(peer = %peer.name, endpoint = %peer.endpoint, err = %e, "Session attempt ended"); + e.to_string() + } + }; + + registry.set_dial_state( + &peer.name, + DialState::Backoff { until: SystemTime::now() + backoff, last_error: reason }, + ); + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(MAX_BACKOFF); + } +} diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index 52189a4..0fb582a 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -33,6 +33,11 @@ pub struct SessionConfig { pub hold_time_secs: u16, pub adj_rib_in: AdjRibIn, pub registry: SessionRegistry, + /// 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 + /// refused with a Bad Peer AS notification. + pub expected_asn: Option, } /// Drive the BGP FSM for a single accepted TCP connection. @@ -112,6 +117,13 @@ pub async fn run_session( peer_open.asn as u32 }; + if let Some(expected) = config.expected_asn + && peer_asn != expected + { + send_notification(&mut writer, BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs), &ctx).await; + bail!("Peer {peer_addr} announced AS{peer_asn} but AS{expected} was configured"); + } + info!( %peer_addr, peer_asn, @@ -152,6 +164,7 @@ pub async fn run_session( let (outbound_tx, mut outbound_rx) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY); let _guard = config.registry.register(SessionHandle::new( peer_id, + config.peer_name.clone(), local_addr, peer_four_byte_asn, outbound_tx, diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index ac05a0b..9af137f 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -1,6 +1,7 @@ pub mod advertise; pub mod config; pub mod connection; +pub mod dialer; pub mod fsm; pub mod metrics; pub mod server; diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index 5aa6194..3032658 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -12,8 +12,9 @@ use bgp_rib::{AdjRibIn, LocalRib}; use crate::advertise::spawn_advertiser; use crate::config::Config; use crate::connection::handle_connection; +use crate::dialer::spawn_dialers; use crate::fsm::SessionConfig; -use crate::session::SessionRegistry; +use crate::session::{ConfiguredPeer, SessionRegistry}; pub struct BgpServerBuilder { config: Config, @@ -55,7 +56,13 @@ impl BgpServerBuilder { listeners, adj_rib_in: AdjRibIn::spawn(), local_rib: LocalRib::spawn(self.config.local_route_policy.clone()), - registry: SessionRegistry::new(), + registry: SessionRegistry::with_configured( + self.config + .peers + .iter() + .map(|p| ConfiguredPeer { name: p.name.clone(), endpoint: p.endpoint, remote_asn: p.remote_asn }) + .collect(), + ), config: self.config, }) } @@ -85,19 +92,22 @@ impl BgpServer { "BGP server started" ); - let session_config = Arc::new(SessionConfig { + let session_config = 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(), registry: self.registry.clone(), - }); + peer_name: None, + expected_asn: None, + }; let _advertiser = spawn_advertiser( self.local_rib.clone(), self.registry.clone(), self.config.local_asn, ); + let _dialers = spawn_dialers(&self.config.peers, &session_config, self.config.connect_retry); let mut accepting: FuturesUnordered<_> = self .listeners @@ -108,11 +118,15 @@ impl BgpServer { 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), - )); + // Inbound connections from a configured peer inherit its + // name, expected ASN and hold time. + let mut config = session_config.clone(); + if let Some(peer) = self.config.peers.iter().find(|p| p.endpoint.ip() == peer_addr.ip()) { + config.peer_name = Some(peer.name.clone()); + config.expected_asn = Some(peer.remote_asn); + config.hold_time_secs = peer.hold_time_secs; + } + tokio::spawn(handle_connection(stream, peer_addr, Arc::new(config))); } Err(e) => { error!(%listen_addr, err = %e, "Accept error"); diff --git a/crates/server/src/session.rs b/crates/server/src/session.rs index d093429..aa72d8c 100644 --- a/crates/server/src/session.rs +++ b/crates/server/src/session.rs @@ -6,7 +6,7 @@ //! the registry to enumerate live sessions and queue outbound messages. use std::collections::HashMap; -use std::net::SocketAddr; +use std::net::{IpAddr, SocketAddr}; use std::sync::{Arc, RwLock}; use std::time::SystemTime; @@ -18,10 +18,30 @@ use crate::metrics::SESSIONS_ESTABLISHED; const EVENT_CHANNEL_CAPACITY: usize = 256; +/// Where the dialer is in bringing up a configured peer. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum DialState { + /// Not dialling: the peer connected to us, or the dialer has not started. + Idle, + Connecting, + /// The last attempt failed or the session ended; waiting to retry. + Backoff { until: SystemTime, last_error: String }, +} + +/// A peer from the configuration file, whether or not it is currently up. +#[derive(Debug, Clone)] +pub struct ConfiguredPeer { + pub name: String, + pub endpoint: SocketAddr, + pub remote_asn: u32, +} + /// Handle to one established session. Cloning is cheap. #[derive(Debug, Clone)] pub struct SessionHandle { pub peer: PeerId, + /// Name from the configuration, if this peer is configured. + pub name: Option, /// Our side of the TCP connection; used as the next hop for routes we /// originate to this peer. pub local_addr: SocketAddr, @@ -39,11 +59,12 @@ pub struct SessionClosed(pub PeerId); impl SessionHandle { pub fn new( peer: PeerId, + name: Option, local_addr: SocketAddr, four_octet_asn: bool, outbound: mpsc::Sender, ) -> Self { - Self { peer, local_addr, established_at: SystemTime::now(), four_octet_asn, outbound } + Self { peer, name, local_addr, established_at: SystemTime::now(), four_octet_asn, outbound } } /// Queues a message for transmission on this session. Waits for queue @@ -62,6 +83,8 @@ pub enum SessionEvent { #[derive(Clone)] pub struct SessionRegistry { inner: Arc>>, + configured: Arc>, + dial_states: Arc>>, events: broadcast::Sender, } @@ -73,8 +96,48 @@ impl Default for SessionRegistry { impl SessionRegistry { pub fn new() -> Self { + Self::with_configured(vec![]) + } + + pub fn with_configured(configured: Vec) -> Self { let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY); - Self { inner: Arc::new(RwLock::new(HashMap::new())), events } + let dial_states = configured.iter().map(|p| (p.name.clone(), DialState::Idle)).collect(); + Self { + inner: Arc::new(RwLock::new(HashMap::new())), + configured: Arc::new(configured), + dial_states: Arc::new(RwLock::new(dial_states)), + events, + } + } + + /// Peers from the configuration file, each with the dialer's current + /// state for it. + pub fn configured_peers(&self) -> Vec<(ConfiguredPeer, DialState)> { + let states = self.dial_states.read().expect("dial states poisoned"); + self.configured + .iter() + .map(|p| (p.clone(), states.get(&p.name).cloned().unwrap_or(DialState::Idle))) + .collect() + } + + /// The configured peer whose endpoint IP is `ip`, if any. + pub fn configured_by_ip(&self, ip: IpAddr) -> Option<&ConfiguredPeer> { + self.configured.iter().find(|p| p.endpoint.ip() == ip) + } + + pub fn set_dial_state(&self, name: &str, state: DialState) { + self.dial_states.write().expect("dial states poisoned").insert(name.to_string(), state); + } + + /// An established session with this remote IP and ASN, regardless of + /// which side opened the TCP connection. + pub fn find(&self, ip: IpAddr, asn: u32) -> Option { + self.inner + .read() + .expect("session registry poisoned") + .values() + .find(|h| h.peer.addr.ip() == ip && h.peer.asn == asn) + .cloned() } /// Adds a session and emits [`SessionEvent::Up`]. The returned guard diff --git a/crates/web/src/api.rs b/crates/web/src/api.rs index bf1baf3..a2c520a 100644 --- a/crates/web/src/api.rs +++ b/crates/web/src/api.rs @@ -11,7 +11,7 @@ use bgp_primitives::ip_prefix::IpPrefix; use bgp_primitives::parser::ParserContext; use bgp_rib::{LocalRibError, LocalRoute, LocalRouteAttrs, SnapshotEntry}; use bgp_server::advertise::build_announce; -use bgp_server::session::SessionHandle; +use bgp_server::session::{ConfiguredPeer, DialState, SessionHandle, SessionRegistry}; use bytes::BytesMut; use serde::{Deserialize, Serialize}; use serde_json::json; @@ -77,27 +77,85 @@ pub struct StatusView { #[derive(Debug, Clone, Serialize)] pub struct PeerView { + /// Name from the configuration file; absent for unconfigured inbound peers. + pub name: Option, pub peer_addr: String, pub peer_asn: u32, - pub local_addr: String, - pub established_at_unix: u64, - pub uptime_secs: u64, - pub four_octet_asn: bool, + /// `Established`, `Connecting`, `Backoff` or `Idle`. + pub state: String, + pub local_addr: Option, + pub established_at_unix: Option, + pub uptime_secs: Option, + pub four_octet_asn: Option, + /// Why the last connection attempt or session ended, while in Backoff. + pub last_error: Option, + pub retry_at_unix: Option, } impl From<&SessionHandle> for PeerView { fn from(s: &SessionHandle) -> Self { Self { + name: s.name.clone(), peer_addr: s.peer.addr.to_string(), peer_asn: s.peer.asn, - local_addr: s.local_addr.to_string(), - established_at_unix: unix_secs(s.established_at), - uptime_secs: s.established_at.elapsed().map(|d| d.as_secs()).unwrap_or(0), - four_octet_asn: s.four_octet_asn, + state: "Established".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)), + four_octet_asn: Some(s.four_octet_asn), + last_error: None, + retry_at_unix: None, } } } +impl PeerView { + fn from_configured(peer: &ConfiguredPeer, state: &DialState) -> Self { + let (state_name, last_error, retry_at_unix) = match state { + DialState::Idle => ("Idle", None, None), + DialState::Connecting => ("Connecting", None, None), + DialState::Backoff { until, last_error } => ("Backoff", Some(last_error.clone()), Some(unix_secs(*until))), + }; + Self { + name: Some(peer.name.clone()), + peer_addr: peer.endpoint.to_string(), + peer_asn: peer.remote_asn, + state: state_name.into(), + local_addr: None, + established_at_unix: None, + uptime_secs: None, + four_octet_asn: None, + last_error, + retry_at_unix, + } + } +} + +/// Configured peers first (with their live session when established), then +/// any unconfigured inbound sessions. +pub fn peer_views(registry: &SessionRegistry) -> Vec { + let sessions = registry.list(); + let mut claimed = std::collections::HashSet::new(); + let mut views: Vec = registry + .configured_peers() + .iter() + .map(|(peer, state)| { + match sessions.iter().find(|s| s.peer.addr.ip() == peer.endpoint.ip() && s.peer.asn == peer.remote_asn) { + Some(session) => { + claimed.insert(session.peer); + PeerView { name: Some(peer.name.clone()), ..PeerView::from(session) } + } + None => PeerView::from_configured(peer, state), + } + }) + .collect(); + let mut rest: Vec = + sessions.iter().filter(|s| !claimed.contains(&s.peer)).map(PeerView::from).collect(); + rest.sort_by(|a, b| a.peer_addr.cmp(&b.peer_addr)); + views.extend(rest); + views +} + #[derive(Debug, Clone, Serialize)] pub struct LocalRouteView { pub prefix: String, @@ -281,9 +339,7 @@ pub async fn status(State(state): State) -> Json { } pub async fn peers(State(state): State) -> Json> { - let mut peers: Vec = state.registry.list().iter().map(PeerView::from).collect(); - peers.sort_by(|a, b| a.peer_addr.cmp(&b.peer_addr)); - Json(peers) + Json(peer_views(&state.registry)) } pub async fn local_routes(State(state): State) -> Json> { diff --git a/crates/web/static/statusz.html b/crates/web/static/statusz.html index f58ead1..95c5a2e 100644 --- a/crates/web/static/statusz.html +++ b/crates/web/static/statusz.html @@ -82,7 +82,7 @@

Peers

- +
peerASNstatelocal address (next hop)establisheduptime4-byte ASN
namepeerASNstatelocal address (next hop)establisheduptime4-byte ASNdetail
@@ -163,16 +163,22 @@ const s = state.status; if (!s) return; $('s-router-id').textContent = s.router_id; $('s-asn').textContent = s.local_asn; $('s-version').textContent = s.version; $('s-uptime').textContent = fmtDur(s.uptime_secs + (now() - s._at)); - $('s-sessions').textContent = state.peers.length; $('s-local').textContent = state.local.length; + $('s-sessions').textContent = state.peers.filter(p => p.state === 'Established').length; $('s-local').textContent = state.local.length; $('s-received').textContent = state.received.length; } + const STATE_CLASS = { Established: 'ok', Connecting: 'warn', Backoff: 'err', Idle: 'muted' }; function renderPeers() { - const b = $('peers-body'); $('peers-count').textContent = state.peers.length; - if (!state.peers.length) { b.innerHTML = 'No established sessions. Peers must connect to this router.'; return; } - b.innerHTML = state.peers.map(p => ` - ${esc(p.peer_addr)}${p.peer_asn}Established${esc(p.local_addr)} - ${fmtTime(p.established_at_unix)}${age(p.established_at_unix)}${p.four_octet_asn ? 'yes' : 'no'}`).join(''); + 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; } + 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) : '—'} + ${up ? fmtTime(p.established_at_unix) : '—'}${up ? age(p.established_at_unix) : '—'}${up ? (p.four_octet_asn ? 'yes' : 'no') : '—'} + ${detail}`; }).join(''); } function renderLocal() { @@ -247,7 +253,7 @@ } function describe(ev) { switch (ev.type) { - case 'session_up': return [`session UP with ${esc(ev.peer.peer_addr)} AS${ev.peer.peer_asn}`, 'ok']; + case 'session_up': return [`session UP with ${esc(ev.peer.name || ev.peer.peer_addr)} AS${ev.peer.peer_asn}`, 'ok']; case 'session_down': return [`session DOWN with ${esc(ev.peer_addr)} AS${ev.peer_asn}`, 'err']; case 'local_announced': return [`announced ${esc(ev.route.prefix)} by ${esc(ev.route.owner)} [${ev.route.communities.map(c => state.wellKnown[c] || c).map(esc).join(' ')}]`, '']; case 'local_withdrawn': return [`withdrew ${esc(ev.prefix)}`, 'warn']; diff --git a/docs/RFCs.md b/docs/RFCs.md index 3f69462..c65bcca 100644 --- a/docs/RFCs.md +++ b/docs/RFCs.md @@ -25,7 +25,9 @@ The base protocol specification. | §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 | ❌ | +| §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) | ❌ | | §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/testdata/server.textpb b/testdata/server.textpb index dccffeb..c9b81dc 100644 --- a/testdata/server.textpb +++ b/testdata/server.textpb @@ -8,6 +8,13 @@ listen_endpoints { address: "::1" port: 179 } +peers { + name: "lab-r1" + endpoint { address: "127.0.0.1" port: 1791 } + remote_asn: 65001 + hold_time_seconds: 30 +} +connect_retry_seconds: 5 local_route_policy { allowed_prefixes: "10.100.0.0/16" allowed_prefixes: "2001:db8:100::/48"