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 {