From a4a180e94844d54d1916e4ceee2d659eca312618 Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Thu, 1 Oct 2026 21:59:55 +0200 Subject: [PATCH] Make the session state machine explicit and free of I/O A session was one function that walked OpenSent, OpenConfirm and Established with the socket, the timers, collision detection, the registry and the RIB all in the same loop. There was no state or event type, and no way to test a transition without a TCP connection. It is now two pieces: - fsm.rs is the state machine of RFC 4271 section 8 for a connection. It takes an Event and returns the Actions to carry out. It owns no socket, timer or task. What the OPENs negotiated lives in the states that have it, so there is nothing to unwrap. Thirteen tests cover the handshake, hold-time negotiation, refused OPENs, messages out of turn, timer expiry, collisions and shutdown by feeding it events. - connection.rs is the driver: it turns the socket, the timers and the outbound queue into events and carries the actions out. Behaviour changes: - A message that arrived in pieces was lost if a timer fired between the pieces: read_exact was used inside select!, which discards what it has read when another branch wins. Messages are now framed from a buffer that survives cancellation. - A NOTIFICATION from the peer during the handshake is no longer answered with an FSM error; RFC 4271 says to close without replying. - A session the peer ended is reported as 'peer sent NOTIFICATION: ...' with its error, where the dialer used to record 'session closed by peer'. Co-Authored-By: Claude Fable 5.1 --- .../tests/basic_connection.rs | 43 + crates/server/src/connection.rs | 481 +++++++- crates/server/src/dialer.rs | 2 +- crates/server/src/error.rs | 39 +- crates/server/src/fsm.rs | 1063 ++++++++++------- crates/server/src/server.rs | 3 +- 6 files changed, 1187 insertions(+), 444 deletions(-) diff --git a/crates/integration-tests/tests/basic_connection.rs b/crates/integration-tests/tests/basic_connection.rs index 8e91bb8..a83e976 100644 --- a/crates/integration-tests/tests/basic_connection.rs +++ b/crates/integration-tests/tests/basic_connection.rs @@ -180,6 +180,49 @@ async fn test_malformed_messages_are_answered_with_the_matching_notification() - Ok(()) } +/// A message may arrive in pieces, with the server's own timers firing in +/// between. The server once threw away the piece it had already read when +/// that happened, so the message was silently lost (and the stream was only +/// back in step if the remainder happened to be thrown away as well). +#[tokio::test] +async fn test_message_split_across_a_keepalive_tick_is_still_one_message() -> Result<()> { + // Hold time 3 s: the server sends a KEEPALIVE every second. + 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?; + + let prefix = IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?; + let mut update = bytes::BytesMut::new(); + Message::Update(UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![], + announced_routes: vec![prefix.clone()], + }) + .to_wire(&peer.ctx, &mut update)?; + + // The first half, a pause long enough for the server's keepalive timer + // to fire, then the rest. + peer.send_raw(&update[..10]).await?; + tokio::time::sleep(Duration::from_millis(1300)).await; + peer.send_raw(&update[10..]).await?; + + tokio::time::timeout(TIMEOUT, async { + while !server + .adj_rib_in + .snapshot() + .await + .iter() + .any(|e| e.prefix == prefix) + { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .map_err(|_| eyre::eyre!("the split UPDATE never reached the RIB"))?; + assert_eq!(server.registry.list().len(), 1, "session was dropped"); + Ok(()) +} + // --------------------------------------------------------------------------- // OpenConfirm → Established // --------------------------------------------------------------------------- diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs index fe9690c..5d4e9ab 100644 --- a/crates/server/src/connection.rs +++ b/crates/server/src/connection.rs @@ -1,11 +1,61 @@ -use std::net::SocketAddr; +//! Drives one TCP connection through the BGP state machine. +//! +//! [`crate::fsm::Fsm`] decides; this module does. It turns what arrives on +//! the socket, the timers and the rest of the server into +//! [`Event`]s, and carries out the [`Action`]s that come back: writing to +//! the socket, arming timers, registering the session, feeding the RIB. + +use std::collections::VecDeque; +use std::future::pending; +use std::io; +use std::net::{Ipv4Addr, SocketAddr}; +use std::pin::Pin; use std::sync::Arc; +use bgp_packet::errors::{BgpError, DecodeError, MsgHeaderSubcode}; +use bgp_packet::message::{BGP_HEADER_LEN, BGP_MAX_MSG_LEN, Message, PathAttribute, UpdateMessage}; +use bgp_primitives::parser::ParserContext; +use bgp_rib::{AdjRibIn, PeerId, ReceivedUpdate}; +use bytes::BytesMut; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; -use tracing::warn; +use tokio::net::tcp::OwnedWriteHalf; +use tokio::sync::mpsc; +use tokio::time::{Instant, Interval, Sleep, interval_at, sleep}; +use tracing::{info, warn}; -use crate::fsm::{SessionConfig, run_session}; +use crate::collision::{CollisionTable, ConnectionGuard, Direction}; +use crate::error::SessionError; +use crate::fsm::{Action, Event, Fsm, FsmConfig, Negotiated}; +use crate::metrics::{ + MESSAGES_RECEIVED, MESSAGES_SENT, SESSIONS_ESTABLISHED_TOTAL, message_type_name, +}; +use crate::session::{SessionGuard, SessionHandle, SessionRegistry}; +/// Number of outbound messages that may be queued for a session before +/// senders are back-pressured. +const OUTBOUND_QUEUE_CAPACITY: usize = 1024; + +/// What a session needs from the rest of the server. +#[derive(Clone)] +pub struct SessionConfig { + pub router_id: Ipv4Addr, + pub local_asn: u32, + 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 + /// refused with a Bad Peer AS notification. + pub expected_asn: Option, +} + +/// Runs a session on an accepted connection and logs how it ended. pub async fn handle_connection( stream: TcpStream, peer_addr: SocketAddr, @@ -15,3 +65,428 @@ pub async fn handle_connection( warn!(%peer_addr, err = %e, "BGP session terminated with error"); } } + +/// Runs the BGP session on `stream` until it ends. +pub async fn run_session( + stream: TcpStream, + peer_addr: SocketAddr, + config: SessionConfig, +) -> Result<(), SessionError> { + let local_addr = stream.local_addr()?; + let (reader, writer) = stream.into_split(); + let (fsm, first_actions) = Fsm::new(FsmConfig { + router_id: config.router_id, + local_asn: config.local_asn, + hold_time_secs: config.hold_time_secs, + expected_asn: config.expected_asn, + }); + let mut session = Session { + fsm, + frames: FrameReader::new(reader), + writer, + ctx: ParserContext::default(), + hold_timer: None, + keepalive_timer: None, + collision: None, + established: None, + peer_addr, + local_addr, + config, + }; + + let result = session.run(first_actions).await; + + // Whatever ended the session, routes learned over it are gone + // (RFC 4271 §8.2.2). Do this before the registration drops so + // subscribers see the withdrawals before the Down event. + if let Some(established) = session.established.take() { + let removed = session + .config + .adj_rib_in + .remove_peer(established.peer_id) + .await; + info!(peer = %established.peer_id, removed_routes = removed, "Session ended"); + } + result +} + +/// What exists only once a session is Established. +struct Established { + peer_id: PeerId, + /// Messages the rest of the server queued for the peer. + outbound: mpsc::Receiver, + /// Keeps the session listed in the registry. + _registration: SessionGuard, +} + +struct Session { + fsm: Fsm, + frames: FrameReader, + writer: OwnedWriteHalf, + /// How messages are encoded and decoded: the width of AS numbers is + /// known once the OPENs are exchanged. + ctx: ParserContext, + hold_timer: Option>>, + keepalive_timer: Option, + /// This connection's slot in the collision table, once admitted. + collision: Option, + established: Option, + peer_addr: SocketAddr, + local_addr: SocketAddr, + config: SessionConfig, +} + +impl Session { + async fn run(&mut self, first_actions: Vec) -> Result<(), SessionError> { + let mut actions = first_actions; + loop { + if let Some(result) = self.execute(actions).await { + return result; + } + let event = self.next_event().await?; + actions = self.fsm.handle(event); + } + } + + /// Waits for whatever happens next. Every branch is cancel-safe: the + /// branches that lose leave nothing half done. + async fn next_event(&mut self) -> Result { + tokio::select! { + frame = self.frames.next(&self.ctx) => match frame { + Ok(msg) => Ok(Event::Message(msg)), + Err(SessionError::Decode(e)) => Ok(Event::DecodeFailed(e)), + Err(e) => Err(e), + }, + () = expired(&mut self.hold_timer) => Ok(Event::HoldTimerExpired), + () = ticked(&mut self.keepalive_timer) => Ok(Event::KeepaliveTimerExpired), + msg = queued(&mut self.established) => Ok(Event::Outbound(msg)), + () = killed(&mut self.collision) => Ok(Event::CollisionLost), + } + } + + /// Carries out `actions` in order. Returns the session's result once it + /// is over. + async fn execute(&mut self, actions: Vec) -> Option> { + let mut queue: VecDeque = actions.into(); + while let Some(action) = queue.pop_front() { + match action { + Action::Send(msg) => { + let closing = matches!(msg, Message::Notification(_)); + if let Err(e) = self.send(&msg).await { + if !closing { + return Some(Err(e)); + } + // The session is ending anyway and the Close that + // follows says why; a NOTIFICATION is best effort. + warn!(peer_addr = %self.peer_addr, err = %e, "Failed to send NOTIFICATION"); + } + } + Action::ResetHoldTimer(duration) => match &mut self.hold_timer { + Some(timer) => timer.as_mut().reset(Instant::now() + duration), + None => self.hold_timer = Some(Box::pin(sleep(duration))), + }, + Action::StopHoldTimer => self.hold_timer = None, + Action::StartKeepaliveTimer(every) => { + // interval_at, so that the first tick is one interval + // from now and not immediate. + self.keepalive_timer = Some(interval_at(Instant::now() + every, every)); + } + Action::Admit(negotiated) => { + self.ctx.four_octet_asn = Some(negotiated.four_octet_asn); + match self.config.collisions.admit( + self.config.router_id, + negotiated.peer_identifier, + self.config.direction, + ) { + Some(guard) => { + self.collision = Some(guard); + info!( + peer_addr = %self.peer_addr, + peer_asn = negotiated.peer_asn, + negotiated_hold_time = negotiated.hold_time_secs, + direction = self.config.direction.as_str(), + "OPEN received, transitioning to OpenConfirm" + ); + } + None => { + // Not admitted: what the machine wanted next no + // longer applies. + info!(peer_addr = %self.peer_addr, "Closing connection: lost collision resolution"); + queue = self.fsm.handle(Event::CollisionLost).into(); + } + } + } + Action::Established(negotiated) => self.establish(negotiated), + Action::DeliverUpdate(update) => { + if let Some(established) = &self.established { + let (withdrawn, announced) = prefix_counts(&update); + info!(peer = %established.peer_id, withdrawn, announced, "Received UPDATE"); + self.config + .adj_rib_in + .apply(established.peer_id, ReceivedUpdate::from_message(&update)) + .await; + } + } + Action::RouteRefreshRequested => { + if let Some(established) = &self.established { + info!(peer = %established.peer_id, "Received ROUTE-REFRESH, re-sending our routes"); + self.config.registry.request_refresh(&established.peer_id); + } + } + Action::Close(result) => return Some(result), + } + } + None + } + + /// Makes the session visible to the rest of the server. + fn establish(&mut self, negotiated: Negotiated) { + let peer_id = PeerId { + addr: self.peer_addr, + asn: negotiated.peer_asn, + }; + info!(peer_addr = %self.peer_addr, peer_asn = negotiated.peer_asn, "Session established"); + SESSIONS_ESTABLISHED_TOTAL.inc(); + + if let Some(collision) = &mut self.collision { + collision.set_established(); + } + let (outbound_tx, outbound) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY); + let registration = self.config.registry.register(SessionHandle::new( + peer_id, + self.config.peer_name.clone(), + self.config.direction, + self.local_addr, + negotiated.four_octet_asn, + outbound_tx, + )); + self.established = Some(Established { + peer_id, + outbound, + _registration: registration, + }); + } + + async fn send(&mut self, msg: &Message) -> Result<(), SessionError> { + if let (Message::Update(update), Some(established)) = (msg, &self.established) { + let (withdrawn, announced) = prefix_counts(update); + info!(peer = %established.peer_id, withdrawn, announced, "Sending UPDATE"); + } + if let Message::Notification(notification) = msg { + info!(peer_addr = %self.peer_addr, error = %notification.error, "Sending NOTIFICATION, closing session"); + } + let mut buf = BytesMut::new(); + msg.to_wire(&self.ctx, &mut buf)?; + self.writer.write_all(&buf).await?; + MESSAGES_SENT + .with_label_values(&[message_type_name(msg)]) + .inc(); + Ok(()) + } +} + +// --------------------------------------------------------------------------- +// Event sources. Each waits forever when its source does not exist yet, so +// that it can sit in the select! from the start. +// --------------------------------------------------------------------------- + +async fn expired(timer: &mut Option>>) { + match timer { + Some(timer) => timer.await, + None => pending().await, + } +} + +async fn ticked(timer: &mut Option) { + match timer { + Some(timer) => { + timer.tick().await; + } + None => pending().await, + } +} + +async fn queued(established: &mut Option) -> Message { + match established { + // The registry holds the sending half for as long as the session is + // registered, so the queue does not close under us; if it somehow + // did, there is simply nothing more to send. + Some(established) => match established.outbound.recv().await { + Some(msg) => msg, + None => pending().await, + }, + None => pending().await, + } +} + +async fn killed(collision: &mut Option) { + match collision { + Some(guard) => { + // An error means the collision table is gone, i.e. the server is + // going away; ending the session is right then too. + let _ = (&mut guard.killed).await; + } + None => pending().await, + } +} + +// --------------------------------------------------------------------------- +// Framing +// --------------------------------------------------------------------------- + +/// Reads whole BGP messages from a byte stream. +/// +/// Bytes are collected in a buffer until a full message is there, so +/// [`FrameReader::next`] can be dropped at any point and called again +/// without losing data. That matters because it shares a `select!` with the +/// timers: a message that arrives in two pieces must survive a keepalive +/// timer firing in between. +struct FrameReader { + reader: R, + buf: BytesMut, +} + +impl FrameReader { + fn new(reader: R) -> Self { + Self { + reader, + buf: BytesMut::with_capacity(usize::from(BGP_MAX_MSG_LEN)), + } + } + + /// The next message from the peer. Cancel-safe. + async fn next(&mut self, ctx: &ParserContext) -> Result { + loop { + if let Some(length) = frame_length(&self.buf)? + && self.buf.len() >= length + { + let frame = self.buf.split_to(length); + let msg = Message::decode(ctx, &frame)?; + MESSAGES_RECEIVED + .with_label_values(&[message_type_name(&msg)]) + .inc(); + return Ok(msg); + } + if self.reader.read_buf(&mut self.buf).await? == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "peer closed the connection", + ) + .into()); + } + } + } +} + +/// The length of the message at the front of `buf`, once its header has +/// arrived. The length decides how much more to wait for, so it is checked +/// here, before the rest of the message exists (RFC 4271 §6.1). +fn frame_length(buf: &[u8]) -> Result, DecodeError> { + if buf.len() < usize::from(BGP_HEADER_LEN) { + return Ok(None); + } + let length = u16::from_be_bytes([buf[16], buf[17]]); + if !(BGP_HEADER_LEN..=BGP_MAX_MSG_LEN).contains(&length) { + return Err(DecodeError { + notification: BgpError::MsgHeader(MsgHeaderSubcode::BadMessageLength), + data: length.to_be_bytes().to_vec(), + detail: format!( + "BGP message length {length} is not between {BGP_HEADER_LEN} and {BGP_MAX_MSG_LEN}" + ), + }); + } + Ok(Some(usize::from(length))) +} + +/// Prefixes withdrawn and announced by an UPDATE, across the IPv4 fields in +/// the message body and the multiprotocol attributes (where IPv6 travels). +fn prefix_counts(update: &UpdateMessage) -> (usize, usize) { + let mut withdrawn = update.withdrawn_routes.len(); + let mut announced = update.announced_routes.len(); + for attr in &update.path_attributes { + match attr { + PathAttribute::MpReachNlri(mp) => announced += mp.prefixes.len(), + PathAttribute::MpUnreachNlri(mp) => withdrawn += mp.prefixes.len(), + _ => {} + } + } + (withdrawn, announced) +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use tokio::io::{AsyncWriteExt, duplex}; + + use super::*; + + const KEEPALIVE: [u8; 19] = [ + 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0xff, 0x00, 0x13, 0x04, + ]; + + #[tokio::test] + async fn messages_are_split_out_of_the_stream_however_they_arrive() { + let (mut peer, ours) = duplex(4096); + let mut frames = FrameReader::new(ours); + let ctx = ParserContext::default(); + + // Two and a half messages in one write, the rest in another. + let mut bytes = Vec::new(); + for _ in 0..3 { + bytes.extend_from_slice(&KEEPALIVE); + } + peer.write_all(&bytes[..47]).await.unwrap(); + assert!(matches!(frames.next(&ctx).await, Ok(Message::KeepAlive))); + assert!(matches!(frames.next(&ctx).await, Ok(Message::KeepAlive))); + peer.write_all(&bytes[47..]).await.unwrap(); + assert!(matches!(frames.next(&ctx).await, Ok(Message::KeepAlive))); + + drop(peer); + assert!(matches!( + frames.next(&ctx).await, + Err(SessionError::Io(e)) if e.kind() == io::ErrorKind::UnexpectedEof + )); + } + + /// The old reader used read_exact inside select!: when a timer fired + /// while a message was half read, the half was thrown away and the + /// stream was out of step from then on. + #[tokio::test] + async fn a_read_abandoned_half_way_loses_nothing() { + let (mut peer, ours) = duplex(4096); + let mut frames = FrameReader::new(ours); + let ctx = ParserContext::default(); + + peer.write_all(&KEEPALIVE[..10]).await.unwrap(); + // Something else wins the race, as a timer would in the session. + let abandoned = tokio::time::timeout(Duration::from_millis(20), frames.next(&ctx)).await; + assert!(abandoned.is_err(), "half a message is not a message"); + + peer.write_all(&KEEPALIVE[10..]).await.unwrap(); + assert!(matches!(frames.next(&ctx).await, Ok(Message::KeepAlive))); + } + + #[tokio::test] + async fn an_impossible_length_is_rejected_from_the_header_alone() { + let (mut peer, ours) = duplex(4096); + let mut frames = FrameReader::new(ours); + + let mut header = KEEPALIVE; + header[16..18].copy_from_slice(&5000u16.to_be_bytes()); + peer.write_all(&header).await.unwrap(); + + // No waiting for 5000 bytes that should never be sent. + match frames.next(&ParserContext::default()).await { + Err(SessionError::Decode(e)) => { + assert_eq!( + e.notification, + BgpError::MsgHeader(MsgHeaderSubcode::BadMessageLength) + ); + assert_eq!(e.data, 5000u16.to_be_bytes()); + } + other => panic!("expected a decode error, got {other:?}"), + } + } +} diff --git a/crates/server/src/dialer.rs b/crates/server/src/dialer.rs index 376648f..619eb49 100644 --- a/crates/server/src/dialer.rs +++ b/crates/server/src/dialer.rs @@ -17,7 +17,7 @@ use tracing::{info, warn}; use crate::collision::Direction; use crate::config::PeerConfig; -use crate::fsm::{SessionConfig, run_session}; +use crate::connection::{SessionConfig, run_session}; use crate::session::{DialState, SessionRegistry}; /// Upper bound for exponential backoff between connection attempts. diff --git a/crates/server/src/error.rs b/crates/server/src/error.rs index d6b4549..fc68fd6 100644 --- a/crates/server/src/error.rs +++ b/crates/server/src/error.rs @@ -8,23 +8,7 @@ use bgp_packet::message::NotificationMessage; use bgp_primitives::parser::EncodeError; use thiserror::Error; -/// The states of a session that has a TCP connection (RFC 4271 §8.2.2). -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SessionState { - OpenSent, - OpenConfirm, - Established, -} - -impl std::fmt::Display for SessionState { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str(match self { - Self::OpenSent => "OpenSent", - Self::OpenConfirm => "OpenConfirm", - Self::Established => "Established", - }) - } -} +use crate::fsm::State; /// Why a session ended abnormally. #[derive(Debug, Error)] @@ -37,14 +21,17 @@ pub enum SessionError { Encode(#[from] EncodeError), /// The peer sent nothing for as long as the pre-negotiation hold time. #[error("timed out waiting for a message in the {0} state")] - Timeout(SessionState), + Timeout(State), #[error("hold timer expired")] HoldTimerExpired, #[error("received {received} in the {state} state")] UnexpectedMessage { - state: SessionState, + state: State, received: &'static str, }, + /// The peer ended the session. Nothing is sent in return. + #[error("peer sent NOTIFICATION: {0}")] + NotifiedByPeer(BgpError), #[error("refused the peer's OPEN: {0}")] OpenRefused(BgpError), #[error("peer announced AS{actual} but AS{expected} was configured")] @@ -58,7 +45,7 @@ impl SessionError { /// or `None` when there is nothing to say or no way to say it. pub fn notification(&self) -> Option { let error = match self { - Self::Io(_) | Self::Encode(_) => return None, + Self::Io(_) | Self::Encode(_) | Self::NotifiedByPeer(_) => return None, Self::Decode(e) => { return Some(NotificationMessage { error: e.notification.clone(), @@ -67,9 +54,11 @@ impl SessionError { } Self::Timeout(_) | Self::HoldTimerExpired => BgpError::HoldTimer, Self::UnexpectedMessage { state, .. } => BgpError::FsmError(match state { - SessionState::OpenSent => FsmSubcode::UnexpectedOpenSent, - SessionState::OpenConfirm => FsmSubcode::UnexpectedOpenConfirm, - SessionState::Established => FsmSubcode::UnexpectedEstablished, + State::OpenSent => FsmSubcode::UnexpectedOpenSent, + State::OpenConfirm => FsmSubcode::UnexpectedOpenConfirm, + // A closed machine handles no messages; Established is the + // only other state one can arrive in. + State::Established | State::Closed => FsmSubcode::UnexpectedEstablished, }), Self::OpenRefused(error) => error.clone(), Self::WrongPeerAs { .. } => BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs), @@ -173,12 +162,12 @@ mod tests { )) ); assert_eq!( - notification(SessionError::Timeout(SessionState::OpenSent)), + notification(SessionError::Timeout(State::OpenSent)), Some((BgpError::HoldTimer, vec![])) ); assert_eq!( notification(SessionError::UnexpectedMessage { - state: SessionState::OpenConfirm, + state: State::OpenConfirm, received: "update" }), Some(( diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index 6526f29..1821759 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -1,454 +1,318 @@ -use std::net::{Ipv4Addr, SocketAddr}; +//! The BGP finite state machine for one connection (RFC 4271 §8), without +//! any I/O. +//! +//! [`Fsm`] is told what happened as an [`Event`] and answers with the +//! [`Action`]s to carry out. It owns no socket, timer or task, so every +//! transition can be tested by calling [`Fsm::handle`]. The driver in +//! [`crate::connection`] turns bytes and timers into events and carries the +//! actions out. +//! +//! Only the states that have a TCP connection live here. Idle, Connect and +//! Active are the dialer's business ([`crate::dialer`]); a connection enters +//! this machine in OpenSent, with our OPEN as its first action. + +use std::net::Ipv4Addr; use std::time::Duration; -use bgp_packet::errors::{BgpError, DecodeError, MsgHeaderSubcode, OpenMsgSubcode}; -use bgp_packet::message::{BGP_HEADER_LEN, BGP_MAX_MSG_LEN, Message, PathAttribute}; +use bgp_packet::errors::{BgpError, DecodeError, OpenMsgSubcode}; +use bgp_packet::message::{AS_TRANS, BGP4_VERSION, Message, UpdateMessage}; use bgp_packet::open::{Capability, OpenMessage, OpenOption}; use bgp_primitives::constants::{AddressFamilyId, SubsequentAfi}; -use bgp_primitives::parser::ParserContext; -use bgp_rib::{AdjRibIn, PeerId, ReceivedUpdate}; -use bytes::BytesMut; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::net::TcpStream; -use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; -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::error::SessionError; +use crate::metrics::message_type_name; -use crate::collision::{CollisionTable, Direction}; -use crate::error::{SessionError, SessionState}; -use crate::metrics::{ - MESSAGES_RECEIVED, MESSAGES_SENT, SESSIONS_ESTABLISHED_TOTAL, message_type_name, -}; -use crate::session::{SessionHandle, SessionRegistry}; +/// How long to wait for the peer's OPEN, and for its first KEEPALIVE when +/// the negotiated hold time is zero. RFC 4271 §8.2.2 suggests four minutes. +pub const INITIAL_HOLD_TIME: Duration = Duration::from_secs(240); -/// Number of outbound messages that may be queued for a session before -/// senders are back-pressured. -const OUTBOUND_QUEUE_CAPACITY: usize = 1024; +/// Where a connection is in the handshake. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum State { + /// Our OPEN is sent; waiting for the peer's. + OpenSent, + /// OPENs are exchanged; waiting for the KEEPALIVE that confirms ours. + OpenConfirm, + /// Routes may be exchanged. + Established, + /// The session is over. Every further event is ignored. + Closed, +} -/// Minimal per-session config extracted from the server config. -#[derive(Clone)] -pub struct SessionConfig { +impl std::fmt::Display for State { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + Self::OpenSent => "OpenSent", + Self::OpenConfirm => "OpenConfirm", + Self::Established => "Established", + Self::Closed => "Closed", + }) + } +} + +/// What the machine needs to know about this end of the session. +#[derive(Debug, Clone)] +pub struct FsmConfig { pub router_id: Ipv4Addr, pub local_asn: u32, + /// Hold time we propose, in seconds. The session uses the smaller of + /// this and the peer's. 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 - /// refused with a Bad Peer AS notification. + /// If set, the AS the peer must announce in its OPEN. pub expected_asn: Option, } -/// 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<(), SessionError> { - let local_addr = stream.local_addr()?; - let (mut reader, mut writer) = stream.into_split(); +/// What the two OPENs settled. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Negotiated { + pub peer_asn: u32, + pub peer_identifier: Ipv4Addr, + /// Zero means the hold and keepalive timers do not run (RFC 4271 §4.2). + pub hold_time_secs: u16, + /// Whether AS numbers are four octets wide on this session (RFC 6793). + pub four_octet_asn: bool, +} - // --- 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?; +/// Something that happened to the connection. +#[derive(Debug)] +pub enum Event { + /// The peer sent a message. + Message(Message), + /// The peer sent something that is not a valid message. + DecodeFailed(DecodeError), + HoldTimerExpired, + KeepaliveTimerExpired, + /// Another connection to the same peer won collision resolution + /// (RFC 4271 §6.8), either when this one asked to be admitted or later. + CollisionLost, + /// The rest of the server wants this message sent to the peer. + Outbound(Message), +} - // RFC 4271 §8.1: use a large initial hold timer (4 minutes) before we - // know the negotiated value. - let pre_open = ParserContext::default(); - let peer_msg = recv_with_timeout( - &mut reader, - &pre_open, - Duration::from_secs(240), - SessionState::OpenSent, - ) - .await; +/// Something the driver must do, in the order given. +#[derive(Debug)] +pub enum Action { + Send(Message), + /// Start the hold timer, or restart it if it is running. + ResetHoldTimer(Duration), + /// The hold timer does not run from here on. + StopHoldTimer, + /// From now on deliver [`Event::KeepaliveTimerExpired`] at this interval. + StartKeepaliveTimer(Duration), + /// The peer's OPEN is acceptable. Ask collision detection whether this + /// connection may go on. If it may not, drop the actions that follow + /// and deliver [`Event::CollisionLost`] instead. + Admit(Negotiated), + /// The session reached Established. + Established(Negotiated), + /// Routes arrived; hand them to the RIB. + DeliverUpdate(UpdateMessage), + /// The peer asked for everything we advertise to be sent again. + RouteRefreshRequested, + /// The session is over, for this reason. Always the last action. + Close(Result<(), SessionError>), +} - let peer_open = match peer_msg { - Ok(Message::Open(o)) => o, - Ok(other) => { - let error = SessionError::UnexpectedMessage { - state: SessionState::OpenSent, - received: message_type_name(&other), - }; - return Err(abort(&mut writer, &pre_open, error).await); - } - Err(e) => return Err(abort(&mut writer, &pre_open, e).await), - }; +/// [`State`] together with what is known in it. What the OPENs settled +/// exists exactly when the handshake has got that far, so it lives in the +/// states that have it and nowhere else. +#[derive(Debug, Clone, Copy)] +enum Phase { + OpenSent, + OpenConfirm(Negotiated), + Established(Negotiated), + Closed, +} - if let Err(refusal) = validate_open(&peer_open, config.router_id) { - let error = SessionError::OpenRefused(refusal); - return Err(abort(&mut writer, &pre_open, error).await); - } +/// The state machine for one connection. +#[derive(Debug)] +pub struct Fsm { + config: FsmConfig, + phase: Phase, +} - 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 - }; - - if let Some(expected) = config.expected_asn - && peer_asn != expected - { - let error = SessionError::WrongPeerAs { - expected, - actual: peer_asn, +impl Fsm { + /// A machine for a connection that has just been opened, and what to do + /// first: send our OPEN and wait for the peer's. + pub fn new(config: FsmConfig) -> (Self, Vec) { + let first = vec![ + Action::Send(build_open(&config)), + Action::ResetHoldTimer(INITIAL_HOLD_TIME), + ]; + let fsm = Self { + config, + phase: Phase::OpenSent, }; - return Err(abort(&mut writer, &ctx, error).await); + (fsm, first) } - // 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?; - - // 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 - }); - - let confirm = tokio::select! { - msg = recv_with_timeout(&mut reader, &ctx, hold_duration, SessionState::OpenConfirm) => msg, - _ = &mut collision_guard.killed => { - return close_for_collision(&mut writer, &ctx, peer_addr).await; + pub fn state(&self) -> State { + match self.phase { + Phase::OpenSent => State::OpenSent, + Phase::OpenConfirm(_) => State::OpenConfirm, + Phase::Established(_) => State::Established, + Phase::Closed => State::Closed, } - }; + } - match confirm { - Ok(Message::KeepAlive) => {} - Ok(other) => { - let error = SessionError::UnexpectedMessage { - state: SessionState::OpenConfirm, + /// What the OPENs settled; `None` until the peer's OPEN is accepted. + pub fn negotiated(&self) -> Option { + match self.phase { + Phase::OpenConfirm(negotiated) | Phase::Established(negotiated) => Some(negotiated), + Phase::OpenSent | Phase::Closed => None, + } + } + + /// Advances the machine by one event. + pub fn handle(&mut self, event: Event) -> Vec { + if matches!(self.phase, Phase::Closed) { + return vec![]; + } + match event { + Event::Message(msg) => self.on_message(msg), + Event::DecodeFailed(e) => self.fail(SessionError::Decode(e)), + Event::CollisionLost => self.fail(SessionError::Collision), + Event::HoldTimerExpired => self.fail(match self.phase { + Phase::Established(_) => SessionError::HoldTimerExpired, + _ => SessionError::Timeout(self.state()), + }), + Event::KeepaliveTimerExpired => match self.phase { + Phase::Established(_) => vec![Action::Send(Message::KeepAlive)], + _ => vec![], + }, + Event::Outbound(msg) => self.on_outbound(msg), + } + } + + fn on_message(&mut self, msg: Message) -> Vec { + match (self.phase, msg) { + // A NOTIFICATION ends the session in any state, and is not + // answered (RFC 4271 §8.2.2). + (_, Message::Notification(notification)) => { + self.close(Err(SessionError::NotifiedByPeer(notification.error))) + } + (Phase::OpenSent, Message::Open(open)) => self.on_open(open), + (Phase::OpenConfirm(negotiated), Message::KeepAlive) => { + self.phase = Phase::Established(negotiated); + let mut actions = vec![Action::Established(negotiated)]; + match hold_time(negotiated) { + Some(hold) => { + actions.push(Action::ResetHoldTimer(hold)); + actions.push(Action::StartKeepaliveTimer(hold / 3)); + } + None => actions.push(Action::StopHoldTimer), + } + actions + } + (Phase::Established(negotiated), Message::KeepAlive) => heard_from_peer(negotiated), + (Phase::Established(negotiated), Message::Update(update)) => { + let mut actions = heard_from_peer(negotiated); + actions.push(Action::DeliverUpdate(update)); + actions + } + (Phase::Established(negotiated), Message::RouteRefresh) => { + let mut actions = heard_from_peer(negotiated); + actions.push(Action::RouteRefreshRequested); + actions + } + (_, other) => self.fail(SessionError::UnexpectedMessage { + state: self.state(), received: message_type_name(&other), - }; - return Err(abort(&mut writer, &ctx, error).await); + }), } - Err(e) => return Err(abort(&mut writer, &ctx, e).await), } - info!(%peer_addr, peer_asn, "Session established"); - SESSIONS_ESTABLISHED_TOTAL.inc(); - - let peer_id = PeerId { - addr: peer_addr, - asn: peer_asn, - }; - - // --- Established -------------------------------------------------------- - // Register so the rest of the server can see this session and queue - // messages for it. The guard deregisters on any exit path. - let (outbound_tx, mut outbound_rx) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY); - 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, - )); - - let result = established_loop( - &mut reader, - &mut writer, - &ctx, - negotiated_hold_time, - peer_id, - &config.adj_rib_in, - SessionInputs { - outbound: &mut outbound_rx, - killed: &mut collision_guard.killed, - registry: &config.registry, - }, - ) - .await; - - // Whatever ended the session, routes learned over it are gone - // (RFC 4271 §8.2.2). Do this before the guard drops so subscribers see - // the withdrawals before the Down event. - let removed = config.adj_rib_in.remove_peer(peer_id).await; - info!(peer = %peer_id, removed_routes = removed, "Session ended"); - 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<()>, - /// Where to report a peer's route-refresh request. - registry: &'a SessionRegistry, -} - -async fn established_loop( - reader: &mut OwnedReadHalf, - writer: &mut OwnedWriteHalf, - ctx: &ParserContext, - hold_time_secs: u16, - peer_id: PeerId, - adj_rib_in: &AdjRibIn, - inputs: SessionInputs<'_>, -) -> Result<(), SessionError> { - let SessionInputs { - outbound, - killed, - registry, - } = 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. - let timers_enabled = hold_time_secs != 0; - let secs = if timers_enabled { - hold_time_secs as u64 - } else { - 3 - }; - let hold_duration = Duration::from_secs(secs); - let keepalive_interval = Duration::from_secs(secs / 3); - - // interval_at avoids firing immediately; first keepalive is sent after one interval. - let mut keepalive_timer = interval_at(Instant::now() + keepalive_interval, keepalive_interval); - - // 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) => { - let msg = match result { - Ok(msg) => msg, - Err(e) => return Err(abort(writer, ctx, e).await), - }; - match msg { - Message::KeepAlive => { - hold_timer.as_mut().reset(Instant::now() + hold_duration); - } - Message::Update(update) => { - hold_timer.as_mut().reset(Instant::now() + hold_duration); - let (withdrawn, announced) = prefix_counts(&update); - info!(peer = %peer_id, withdrawn, announced, "Received UPDATE"); - adj_rib_in.apply(peer_id, ReceivedUpdate::from_message(&update)).await; - } - Message::Notification(notif) => { - info!(peer = %peer_id, error = %notif.error, "Received NOTIFICATION, closing session"); - return Ok(()); - } - Message::Open(_) => { - let error = SessionError::UnexpectedMessage { - state: SessionState::Established, - received: "open", - }; - return Err(abort(writer, ctx, error).await); - } - Message::RouteRefresh => { - hold_timer.as_mut().reset(Instant::now() + hold_duration); - info!(peer = %peer_id, "Received ROUTE-REFRESH, re-sending our routes"); - registry.request_refresh(&peer_id); - } - } - } - Some(msg) = outbound.recv() => { - if let Message::Update(update) = &msg { - let (withdrawn, announced) = prefix_counts(update); - info!(peer = %peer_id, withdrawn, announced, "Sending UPDATE"); - } - send_message(writer, &msg, ctx).await?; - // A NOTIFICATION from our side ends the session (RFC 4271 §6). - if let Message::Notification(notif) = &msg { - info!(peer = %peer_id, error = %notif.error, "Sent NOTIFICATION, closing session"); - return Ok(()); - } - } - _ = &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?; - } - _ = &mut hold_timer, if timers_enabled => { - warn!(peer = %peer_id, "Hold timer expired"); - return Err(abort(writer, ctx, SessionError::HoldTimerExpired).await); - } + fn on_open(&mut self, open: OpenMessage) -> Vec { + if let Err(refusal) = validate_open(&open, self.config.router_id) { + return self.fail(SessionError::OpenRefused(refusal)); } + // A four-octet AS number does not fit the OPEN's own field; a peer + // that has one puts AS_TRANS there and the number in a capability. + let capability_asn = open.options.iter().find_map(|option| match option { + OpenOption::Capabilities(caps) => caps.iter().find_map(|cap| match cap { + Capability::FourByteAsn { asn } => Some(*asn), + _ => None, + }), + }); + let peer_asn = capability_asn.unwrap_or(u32::from(open.asn)); + if let Some(expected) = self.config.expected_asn + && peer_asn != expected + { + return self.fail(SessionError::WrongPeerAs { + expected, + actual: peer_asn, + }); + } + + let negotiated = Negotiated { + peer_asn, + peer_identifier: open.identifier, + hold_time_secs: self.config.hold_time_secs.min(open.hold_time), + four_octet_asn: capability_asn.is_some(), + }; + self.phase = Phase::OpenConfirm(negotiated); + vec![ + Action::Admit(negotiated), + Action::Send(Message::KeepAlive), + // With a hold time of zero the timer will not run once + // Established, but a peer that never confirms must still be + // given up on. + Action::ResetHoldTimer(hold_time(negotiated).unwrap_or(INITIAL_HOLD_TIME)), + ] + } + + fn on_outbound(&mut self, msg: Message) -> Vec { + if !matches!(self.phase, Phase::Established(_)) { + // Nothing may be sent before Established but the handshake. + return vec![]; + } + // A NOTIFICATION from our side ends the session (RFC 4271 §6). + let ends_session = matches!(msg, Message::Notification(_)); + let mut actions = vec![Action::Send(msg)]; + if ends_session { + actions.extend(self.close(Ok(()))); + } + actions + } + + /// Ends the session because of `error`, telling the peer with the + /// NOTIFICATION that goes with it if there is one. + fn fail(&mut self, error: SessionError) -> Vec { + let mut actions: Vec = error + .notification() + .map(|notification| Action::Send(Message::Notification(notification))) + .into_iter() + .collect(); + actions.extend(self.close(Err(error))); + actions + } + + fn close(&mut self, result: Result<(), SessionError>) -> Vec { + self.phase = Phase::Closed; + vec![Action::Close(result)] } } -// --------------------------------------------------------------------------- -// Wire helpers -// --------------------------------------------------------------------------- +/// Any message from the peer shows it is alive (RFC 4271 §4.4). +fn heard_from_peer(negotiated: Negotiated) -> Vec { + hold_time(negotiated) + .map(Action::ResetHoldTimer) + .into_iter() + .collect() +} -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?; - - // The length decides how much more to read, so it is checked before the - // rest of the message exists (RFC 4271 §6.1). - let length = u16::from_be_bytes([header[16], header[17]]); - if !(BGP_HEADER_LEN..=BGP_MAX_MSG_LEN).contains(&length) { - return Err(DecodeError { - notification: BgpError::MsgHeader(MsgHeaderSubcode::BadMessageLength), - data: length.to_be_bytes().to_vec(), - detail: format!( - "BGP message length {length} is not between {BGP_HEADER_LEN} and {BGP_MAX_MSG_LEN}" - ), - } - .into()); +/// The hold time as a duration; `None` when the timers do not run. +fn hold_time(negotiated: Negotiated) -> Option { + match negotiated.hold_time_secs { + 0 => None, + secs => Some(Duration::from_secs(secs.into())), } - let total_len = usize::from(length); - - 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::decode(ctx, &buf)?; - MESSAGES_RECEIVED - .with_label_values(&[message_type_name(&msg)]) - .inc(); - Ok(msg) } -async fn recv_with_timeout( - reader: &mut OwnedReadHalf, - ctx: &ParserContext, - timeout: Duration, - state: SessionState, -) -> Result { - tokio::time::timeout(timeout, read_message(reader, ctx)) - .await - .map_err(|_| SessionError::Timeout(state))? -} - -async fn send_message( - writer: &mut OwnedWriteHalf, - msg: &Message, - ctx: &ParserContext, -) -> Result<(), SessionError> { - let mut buf = BytesMut::new(); - msg.to_wire(ctx, &mut buf)?; - writer.write_all(&buf).await?; - MESSAGES_SENT - .with_label_values(&[message_type_name(msg)]) - .inc(); - Ok(()) -} - -/// Ends the session because of `error`: tells the peer with the NOTIFICATION -/// that goes with it, if there is one, and hands the error back to return. -/// Sending is best effort, since the session is being torn down anyway. -async fn abort( - writer: &mut OwnedWriteHalf, - ctx: &ParserContext, - error: SessionError, -) -> SessionError { - if let Some(notification) = error.notification() { - let msg = Message::Notification(notification); - if let Err(e) = send_message(writer, &msg, ctx).await { - warn!("Failed to send NOTIFICATION: {e}"); - } - } - error -} - -/// 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<(), SessionError> { - info!(%peer_addr, "Closing connection: lost collision resolution"); - Err(abort(writer, ctx, SessionError::Collision).await) -} - -// --------------------------------------------------------------------------- -// RIB helpers -// --------------------------------------------------------------------------- - -/// Prefixes withdrawn and announced by an UPDATE, across the IPv4 fields in -/// the message body and the multiprotocol attributes (where IPv6 travels). -fn prefix_counts(update: &bgp_packet::message::UpdateMessage) -> (usize, usize) { - let mut withdrawn = update.withdrawn_routes.len(); - let mut announced = update.announced_routes.len(); - for attr in &update.path_attributes { - match attr { - PathAttribute::MpReachNlri(mp) => announced += mp.prefixes.len(), - PathAttribute::MpUnreachNlri(mp) => withdrawn += mp.prefixes.len(), - _ => {} - } - } - (withdrawn, announced) -} - -// --------------------------------------------------------------------------- -// OPEN construction and validation -// --------------------------------------------------------------------------- - -fn build_open(config: &SessionConfig) -> Message { +fn build_open(config: &FsmConfig) -> 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 - }; - + let asn_field = u16::try_from(config.local_asn).unwrap_or(AS_TRANS); Message::Open(OpenMessage { version: BGP4_VERSION, asn: asn_field, @@ -487,3 +351,376 @@ fn validate_open(open: &OpenMessage, local_id: Ipv4Addr) -> Result<(), BgpError> } Ok(()) } + +#[cfg(test)] +mod tests { + use bgp_packet::errors::{CeaseSubcode, FsmSubcode, MsgHeaderSubcode}; + use bgp_packet::message::NotificationMessage; + + use super::*; + + const LOCAL_ID: Ipv4Addr = Ipv4Addr::new(10, 0, 0, 1); + const PEER_ID: Ipv4Addr = Ipv4Addr::new(10, 0, 0, 2); + + fn config() -> FsmConfig { + FsmConfig { + router_id: LOCAL_ID, + local_asn: 65000, + hold_time_secs: 90, + expected_asn: None, + } + } + + fn peer_open(hold_time: u16) -> OpenMessage { + OpenMessage { + version: BGP4_VERSION, + asn: 65001, + hold_time, + identifier: PEER_ID, + options: vec![], + } + } + + /// A machine taken as far as `state` by a well-behaved peer. + fn fsm_in(state: State, peer_hold_time: u16) -> Fsm { + let (mut fsm, _) = Fsm::new(config()); + if state != State::OpenSent { + fsm.handle(Event::Message(Message::Open(peer_open(peer_hold_time)))); + } + if state == State::Established { + fsm.handle(Event::Message(Message::KeepAlive)); + } + assert_eq!(fsm.state(), state); + fsm + } + + fn empty_update() -> UpdateMessage { + UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![], + announced_routes: vec![], + } + } + + /// The NOTIFICATION sent and the reason given, for a machine that ended + /// the session. + fn ending(fsm: &Fsm, actions: Vec) -> (Option, Result<(), SessionError>) { + assert_eq!(fsm.state(), State::Closed); + let mut sent = None; + let mut result = None; + let count = actions.len(); + for (i, action) in actions.into_iter().enumerate() { + match action { + Action::Send(Message::Notification(n)) => sent = Some(n.error), + Action::Close(r) => { + assert_eq!(i, count - 1, "Close is the last action"); + result = Some(r); + } + other => panic!("unexpected action while closing: {other:?}"), + } + } + (sent, result.expect("a closing machine says why")) + } + + #[test] + fn a_new_connection_sends_open_and_waits_four_minutes_for_the_peers() { + let (fsm, actions) = Fsm::new(config()); + assert_eq!(fsm.state(), State::OpenSent); + assert!(matches!( + &actions[..], + [ + Action::Send(Message::Open(open)), + Action::ResetHoldTimer(INITIAL_HOLD_TIME) + ] if open.identifier == LOCAL_ID && open.hold_time == 90 + )); + } + + #[test] + fn the_handshake_reaches_established_and_starts_the_timers() { + let (mut fsm, _) = Fsm::new(config()); + + // The peer proposes 30 seconds, we 90: the smaller one holds. + let actions = fsm.handle(Event::Message(Message::Open(peer_open(30)))); + assert_eq!(fsm.state(), State::OpenConfirm); + let expected = Negotiated { + peer_asn: 65001, + peer_identifier: PEER_ID, + hold_time_secs: 30, + four_octet_asn: false, + }; + assert!(matches!( + &actions[..], + [ + Action::Admit(n), + Action::Send(Message::KeepAlive), + Action::ResetHoldTimer(hold) + ] if *n == expected && *hold == Duration::from_secs(30) + )); + + let actions = fsm.handle(Event::Message(Message::KeepAlive)); + assert_eq!(fsm.state(), State::Established); + assert!(matches!( + &actions[..], + [ + Action::Established(n), + Action::ResetHoldTimer(hold), + Action::StartKeepaliveTimer(keepalive) + ] if *n == expected + && *hold == Duration::from_secs(30) + && *keepalive == Duration::from_secs(10) + )); + } + + #[test] + fn a_hold_time_of_zero_runs_no_timers_once_established() { + let (mut fsm, _) = Fsm::new(config()); + let actions = fsm.handle(Event::Message(Message::Open(peer_open(0)))); + // The KEEPALIVE that confirms the session is still waited for. + assert!(matches!( + actions.last(), + Some(Action::ResetHoldTimer(INITIAL_HOLD_TIME)) + )); + + let actions = fsm.handle(Event::Message(Message::KeepAlive)); + assert!(matches!( + &actions[..], + [Action::Established(_), Action::StopHoldTimer] + )); + // And hearing from the peer restarts nothing. + assert!(fsm.handle(Event::Message(Message::KeepAlive)).is_empty()); + } + + #[test] + fn a_four_octet_as_number_comes_from_the_capability() { + let (mut fsm, _) = Fsm::new(FsmConfig { + expected_asn: Some(210036), + ..config() + }); + let open = OpenMessage { + asn: AS_TRANS, + options: vec![OpenOption::Capabilities(vec![Capability::FourByteAsn { + asn: 210036, + }])], + ..peer_open(90) + }; + fsm.handle(Event::Message(Message::Open(open))); + let negotiated = fsm.negotiated().unwrap(); + assert_eq!(negotiated.peer_asn, 210036); + assert!(negotiated.four_octet_asn); + } + + #[test] + fn an_unacceptable_open_is_refused_with_the_matching_subcode() { + let refused = |open: OpenMessage, config: FsmConfig| { + let (mut fsm, _) = Fsm::new(config); + let actions = fsm.handle(Event::Message(Message::Open(open))); + ending(&fsm, actions).0 + }; + let open_error = |subcode| Some(BgpError::OpenMsg(subcode)); + + assert_eq!( + refused( + OpenMessage { + version: 3, + ..peer_open(90) + }, + config() + ), + open_error(OpenMsgSubcode::UnsupportedVersion) + ); + assert_eq!( + refused( + OpenMessage { + identifier: LOCAL_ID, + ..peer_open(90) + }, + config() + ), + open_error(OpenMsgSubcode::BadBgpId), + "a peer cannot have our identifier" + ); + assert_eq!( + refused(peer_open(2), config()), + open_error(OpenMsgSubcode::UnacceptableHoldTime) + ); + assert_eq!( + refused( + peer_open(90), + FsmConfig { + expected_asn: Some(64999), + ..config() + } + ), + open_error(OpenMsgSubcode::BadPeerAs) + ); + } + + #[test] + fn a_message_out_of_turn_is_an_fsm_error_for_the_state_it_arrived_in() { + let cases = [ + ( + State::OpenSent, + Message::KeepAlive, + FsmSubcode::UnexpectedOpenSent, + ), + ( + State::OpenConfirm, + Message::Update(empty_update()), + FsmSubcode::UnexpectedOpenConfirm, + ), + ( + State::Established, + Message::Open(peer_open(90)), + FsmSubcode::UnexpectedEstablished, + ), + ]; + for (state, msg, subcode) in cases { + let mut fsm = fsm_in(state, 90); + let actions = fsm.handle(Event::Message(msg)); + let (sent, result) = ending(&fsm, actions); + assert_eq!(sent, Some(BgpError::FsmError(subcode)), "{state}"); + assert!( + matches!(result, Err(SessionError::UnexpectedMessage { state: s, .. }) if s == state) + ); + } + } + + #[test] + fn silence_ends_the_session_with_hold_timer_expired_in_every_state() { + for state in [State::OpenSent, State::OpenConfirm, State::Established] { + let mut fsm = fsm_in(state, 90); + let actions = fsm.handle(Event::HoldTimerExpired); + let (sent, result) = ending(&fsm, actions); + assert_eq!(sent, Some(BgpError::HoldTimer), "{state}"); + match state { + State::Established => { + assert!(matches!(result, Err(SessionError::HoldTimerExpired))) + } + waiting => { + assert!(matches!(result, Err(SessionError::Timeout(s)) if s == waiting)) + } + } + } + } + + #[test] + fn losing_a_collision_sends_cease_whenever_it_happens() { + for state in [State::OpenConfirm, State::Established] { + let mut fsm = fsm_in(state, 90); + let actions = fsm.handle(Event::CollisionLost); + let (sent, result) = ending(&fsm, actions); + assert_eq!( + sent, + Some(BgpError::Cease(CeaseSubcode::ConnCollisionResolution)), + "{state}" + ); + assert!(matches!(result, Err(SessionError::Collision))); + } + } + + #[test] + fn undecodable_input_is_answered_with_the_notification_the_decoder_named() { + let mut fsm = fsm_in(State::Established, 90); + let actions = fsm.handle(Event::DecodeFailed(DecodeError { + notification: BgpError::MsgHeader(MsgHeaderSubcode::BadMessageType), + data: vec![9], + detail: "unknown BGP message type 9".into(), + })); + assert!(matches!( + &actions[0], + Action::Send(Message::Notification(n)) + if n.error == BgpError::MsgHeader(MsgHeaderSubcode::BadMessageType) && n.data == [9] + )); + assert!(matches!( + ending(&fsm, actions).1, + Err(SessionError::Decode(_)) + )); + } + + #[test] + fn a_notification_from_the_peer_ends_the_session_without_a_reply() { + for state in [State::OpenSent, State::OpenConfirm, State::Established] { + let mut fsm = fsm_in(state, 90); + let cease = BgpError::Cease(CeaseSubcode::AdminShutdown); + let actions = fsm.handle(Event::Message(Message::Notification(NotificationMessage { + error: cease.clone(), + data: vec![], + }))); + let (sent, result) = ending(&fsm, actions); + assert_eq!(sent, None, "{state}"); + assert!(matches!(result, Err(SessionError::NotifiedByPeer(e)) if e == cease)); + } + } + + #[test] + fn established_traffic_restarts_the_hold_timer_and_is_passed_on() { + let mut fsm = fsm_in(State::Established, 30); + let hold = Duration::from_secs(30); + + let actions = fsm.handle(Event::Message(Message::KeepAlive)); + assert!(matches!(&actions[..], [Action::ResetHoldTimer(d)] if *d == hold)); + + let actions = fsm.handle(Event::Message(Message::Update(empty_update()))); + assert!(matches!( + &actions[..], + [Action::ResetHoldTimer(d), Action::DeliverUpdate(_)] if *d == hold + )); + + let actions = fsm.handle(Event::Message(Message::RouteRefresh)); + assert!(matches!( + &actions[..], + [Action::ResetHoldTimer(d), Action::RouteRefreshRequested] if *d == hold + )); + + let actions = fsm.handle(Event::KeepaliveTimerExpired); + assert!(matches!(&actions[..], [Action::Send(Message::KeepAlive)])); + assert_eq!(fsm.state(), State::Established); + } + + #[test] + fn outbound_messages_are_sent_only_once_established() { + let mut waiting = fsm_in(State::OpenConfirm, 90); + assert!( + waiting + .handle(Event::Outbound(Message::Update(empty_update()))) + .is_empty() + ); + assert_eq!(waiting.state(), State::OpenConfirm); + + let mut fsm = fsm_in(State::Established, 90); + let actions = fsm.handle(Event::Outbound(Message::Update(empty_update()))); + assert!(matches!(&actions[..], [Action::Send(Message::Update(_))])); + + // Sending a NOTIFICATION is how the server shuts a session down. + let actions = fsm.handle(Event::Outbound(Message::Notification( + NotificationMessage { + error: BgpError::Cease(CeaseSubcode::AdminShutdown), + data: vec![], + }, + ))); + assert!(matches!( + &actions[..], + [ + Action::Send(Message::Notification(_)), + Action::Close(Ok(())) + ] + )); + assert_eq!(fsm.state(), State::Closed); + } + + #[test] + fn a_closed_machine_ignores_everything() { + let mut fsm = fsm_in(State::Established, 90); + fsm.handle(Event::HoldTimerExpired); + assert_eq!(fsm.state(), State::Closed); + for event in [ + Event::Message(Message::KeepAlive), + Event::HoldTimerExpired, + Event::KeepaliveTimerExpired, + Event::CollisionLost, + Event::Outbound(Message::KeepAlive), + ] { + assert!(fsm.handle(event).is_empty()); + } + } +} diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index 8f8fee1..2262f51 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -11,10 +11,9 @@ 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::connection::{SessionConfig, handle_connection}; use crate::dialer::spawn_dialers; use crate::error::ServerError; -use crate::fsm::SessionConfig; use crate::session::{ConfiguredPeer, SessionRegistry}; pub struct BgpServerBuilder {