From 16a4e3b640f080b675749786febc80b8d33fdb05 Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Thu, 1 Oct 2026 21:18:35 +0200 Subject: [PATCH] Typed errors in the library crates; send the NOTIFICATION RFC 4271 assigns The library crates returned eyre reports, so a failure was a string and the session code could not tell a truncated UPDATE from a bad header. Whatever went wrong, the peer got an FSM error or nothing. Each crate now has its own error types: - bgp-primitives: PrefixError, UnknownAfi and UnknownSafi, and EncodeError for every to_wire. - bgp-packet: Message::decode returns a DecodeError naming the NOTIFICATION code, subcode and data for the failure (header errors per section 6.1, OPEN per 6.2, UPDATE per 6.3). InvalidValue and CommunityError replace the remaining strings. from_wire is now a thin wrapper around decode. - bgp-server: SessionError, ServerError and ConfigError. Every abnormal end of a session goes through one function that sends the NOTIFICATION the error maps to. The message length is checked against the 4096-octet maximum before the body is read. - bgp-web: serve returns io::Error. eyre remains in the binary and in tests. Bugs found and fixed on the way: - An OPEN with no optional parameters, which is valid, failed to parse. - An unsupported BGP version was reported as an unknown optional parameter. It now gets Unsupported Version Number with the version we speak as data. - A SAFI above 255 was truncated to its low octet and could match a known one. - Prefix text with a length beyond the address size (10.0.0.0/99) was accepted. - Message::to_wire truncated the body length to 16 bits before checking it. - A peer hold time above 65535 seconds in the configuration wrapped. Co-Authored-By: Claude Fable 5.1 --- Cargo.lock | 3 - bin/bgp/src/main.rs | 2 +- crates/integration-tests/src/test_server.rs | 5 +- .../tests/basic_connection.rs | 108 +++- crates/packet/Cargo.toml | 4 +- crates/packet/src/community.rs | 37 +- crates/packet/src/errors.rs | 24 + crates/packet/src/message.rs | 481 ++++++++++++------ crates/packet/src/open.rs | 94 ++-- crates/primitives/Cargo.toml | 1 - crates/primitives/src/constants.rs | 64 ++- crates/primitives/src/ip_prefix.rs | 117 +++-- crates/primitives/src/parser.rs | 39 +- crates/server/Cargo.toml | 1 - crates/server/src/config.rs | 284 +++++++---- crates/server/src/dialer.rs | 11 +- crates/server/src/error.rs | 202 ++++++++ crates/server/src/fsm.rs | 153 +++--- crates/server/src/lib.rs | 1 + crates/server/src/server.rs | 12 +- crates/server/src/tracing.rs | 6 +- crates/web/Cargo.toml | 1 - crates/web/src/lib.rs | 14 +- 23 files changed, 1171 insertions(+), 493 deletions(-) create mode 100644 crates/server/src/error.rs diff --git a/Cargo.lock b/Cargo.lock index 138d5c6..9605e44 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -194,7 +194,6 @@ version = "0.0.2" dependencies = [ "bitfield", "bytes", - "eyre", "nom", "serde", "serde_repr", @@ -223,7 +222,6 @@ dependencies = [ "bgp-rib", "bytes", "clap", - "eyre", "futures", "prometheus", "protobuf", @@ -242,7 +240,6 @@ dependencies = [ "bgp-packet", "bgp-rib", "bgp-server", - "eyre", "futures", "prometheus", "serde", diff --git a/bin/bgp/src/main.rs b/bin/bgp/src/main.rs index 8b2a7bc..572969d 100644 --- a/bin/bgp/src/main.rs +++ b/bin/bgp/src/main.rs @@ -58,7 +58,7 @@ async fn main() -> Result<()> { // makes them slower to take us back. let registry = server.registry.clone(); tokio::select! { - result = server.run() => result, + result = server.run() => Ok(result?), signal = shutdown_signal() => { info!(signal, "Shutting down, closing BGP sessions"); let left = registry.shutdown(Duration::from_secs(2)).await; diff --git a/crates/integration-tests/src/test_server.rs b/crates/integration-tests/src/test_server.rs index 06eca38..d0a956a 100644 --- a/crates/integration-tests/src/test_server.rs +++ b/crates/integration-tests/src/test_server.rs @@ -2,6 +2,7 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use bgp_rib::{AdjRibIn, LocalRib}; use bgp_server::config::Config; +use bgp_server::error::ServerError; use bgp_server::server::BgpServerBuilder; use bgp_server::session::SessionRegistry; use bgp_web::{AppState, RouterInfo}; @@ -12,7 +13,7 @@ use tokio::task::JoinHandle; /// A BGP server started on an ephemeral port for testing. /// Aborts the server task when dropped. pub struct TestServer { - handle: JoinHandle>, + handle: JoinHandle>, /// The single IPv4 loopback address the server is bound to. pub addr: SocketAddr, /// Handle to the server's Adj-RIB-In for test introspection. @@ -96,7 +97,7 @@ impl TestServer { /// Aborts the HTTP API task when dropped. pub struct HttpGuard { - handle: JoinHandle>, + handle: JoinHandle>, } impl Drop for HttpGuard { diff --git a/crates/integration-tests/tests/basic_connection.rs b/crates/integration-tests/tests/basic_connection.rs index b3bd139..8e91bb8 100644 --- a/crates/integration-tests/tests/basic_connection.rs +++ b/crates/integration-tests/tests/basic_connection.rs @@ -3,7 +3,7 @@ use std::time::Duration; use bgp_integration_tests::test_peer::TestPeer; use bgp_integration_tests::test_server::TestServer; -use bgp_packet::errors::{BgpError, CeaseSubcode}; +use bgp_packet::errors::{BgpError, CeaseSubcode, FsmSubcode, MsgHeaderSubcode, UpdateMsgSubcode}; use bgp_packet::message::{ Message, MpReachNlriPathAttribute, MpUnreachNlriPathAttribute, NlriNextHop, NotificationMessage, PathAttribute, UpdateMessage, @@ -13,6 +13,8 @@ use bgp_primitives::ip_prefix::IpPrefix; use bgp_rib::RibEvent; use eyre::Result; +const TIMEOUT: Duration = Duration::from_secs(5); + // --------------------------------------------------------------------------- // Connectivity // --------------------------------------------------------------------------- @@ -74,6 +76,110 @@ async fn test_bad_marker_does_not_crash_server() -> Result<()> { Ok(()) } +/// Reads until a NOTIFICATION arrives, skipping KEEPALIVEs. +async fn expect_notification(peer: &mut TestPeer) -> Result { + tokio::time::timeout(TIMEOUT, async { + loop { + match peer.recv().await? { + Message::Notification(n) => return Ok(n), + Message::KeepAlive => {} + other => eyre::bail!("expected NOTIFICATION, got {other:?}"), + } + } + }) + .await? +} + +/// A header with the given length and type, followed by `body`. +fn framed(length: u16, msg_type: u8, body: &[u8]) -> Vec { + let mut buf = vec![0xFF; 16]; + buf.extend_from_slice(&length.to_be_bytes()); + buf.push(msg_type); + buf.extend_from_slice(body); + buf +} + +/// What the peer sent decides the NOTIFICATION it gets back (RFC 4271 §6), +/// in every state, and the session ends. +#[tokio::test] +async fn test_malformed_messages_are_answered_with_the_matching_notification() -> Result<()> { + let server = TestServer::start(65000).await?; + let header = |subcode| BgpError::MsgHeader(subcode); + let mut bad_marker = framed(19, 4, &[]); + bad_marker[0] = 0; + + let cases: Vec<(&str, Vec, BgpError, Vec)> = vec![ + ( + "length beyond the maximum", + framed(5000, 2, &[]), + header(MsgHeaderSubcode::BadMessageLength), + 5000u16.to_be_bytes().to_vec(), + ), + ( + "length below a header", + framed(5, 4, &[]), + header(MsgHeaderSubcode::BadMessageLength), + vec![0, 5], + ), + ( + "unknown message type", + framed(19, 9, &[]), + header(MsgHeaderSubcode::BadMessageType), + vec![9], + ), + ( + "marker not all ones", + bad_marker, + header(MsgHeaderSubcode::ConnNotSynchronized), + vec![], + ), + ( + "UPDATE with attributes running past its end", + framed(23, 2, &[0, 0, 0, 40]), + BgpError::UpdateMsg(UpdateMsgSubcode::MalformedAttrs), + vec![], + ), + ]; + + for (i, (what, bytes, error, data)) in cases.into_iter().enumerate() { + // Established: a distinct router ID each time, so that a session + // still closing does not collide with the next one. + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(65001, Ipv4Addr::new(10, 0, 1, i as u8 + 1), 90) + .await?; + peer.send_raw(&bytes).await?; + let notification = expect_notification(&mut peer).await?; + assert_eq!( + (notification.error, notification.data), + (error.clone(), data.clone()), + "{what}, Established" + ); + + // OpenSent: the same bytes instead of an OPEN. + let mut peer = TestPeer::connect(server.addr).await?; + let _ = peer.recv_open().await?; + peer.send_raw(&bytes).await?; + let notification = expect_notification(&mut peer).await?; + assert_eq!( + (notification.error, notification.data), + (error, data), + "{what}, OpenSent" + ); + } + + // A well-formed message at the wrong time is an FSM error, not a + // message error. + let mut peer = TestPeer::connect(server.addr).await?; + let _ = peer.recv_open().await?; + peer.send(&Message::KeepAlive).await?; + let notification = expect_notification(&mut peer).await?; + assert_eq!( + notification.error, + BgpError::FsmError(FsmSubcode::UnexpectedOpenSent) + ); + Ok(()) +} + // --------------------------------------------------------------------------- // OpenConfirm → Established // --------------------------------------------------------------------------- diff --git a/crates/packet/Cargo.toml b/crates/packet/Cargo.toml index b8d3315..fb4e60b 100644 --- a/crates/packet/Cargo.toml +++ b/crates/packet/Cargo.toml @@ -11,7 +11,6 @@ version.workspace = true [dependencies] bitfield.workspace = true bytes.workspace = true -eyre.workspace = true nom.workspace = true serde.workspace = true serde_repr.workspace = true @@ -19,5 +18,8 @@ thiserror.workspace = true strum.workspace = true bgp-primitives.workspace = true +[dev-dependencies] +eyre.workspace = true + [lints] workspace = true diff --git a/crates/packet/src/community.rs b/crates/packet/src/community.rs index d078feb..aade8ed 100644 --- a/crates/packet/src/community.rs +++ b/crates/packet/src/community.rs @@ -5,8 +5,19 @@ //! of the well-known names from RFC 1997 / RFC 7999 / RFC 8326. Large //! communities are written `GLOBAL:LOCAL1:LOCAL2` (RFC 8092). +use thiserror::Error; + use crate::message::LargeCommunity; +/// Text that is not a community. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum CommunityError { + #[error("community {0:?} must be ASN:VALUE with 16-bit numbers, or a well-known name")] + Standard(String), + #[error("large community {0:?} must be GLOBAL:LOCAL1:LOCAL2 with 32-bit numbers")] + Large(String), +} + /// Well-known standard communities, by name. pub const WELL_KNOWN: &[(&str, (u16, u16))] = &[ ("GRACEFUL_SHUTDOWN", (0xFFFF, 0x0000)), @@ -17,22 +28,17 @@ pub const WELL_KNOWN: &[(&str, (u16, u16))] = &[ ("NOPEER", (0xFFFF, 0xFF04)), ]; -pub fn parse_standard(s: &str) -> Result<(u16, u16), String> { +pub fn parse_standard(s: &str) -> Result<(u16, u16), CommunityError> { let s = s.trim(); + let invalid = || CommunityError::Standard(s.to_string()); if let Some((_, value)) = WELL_KNOWN .iter() .find(|(name, _)| name.eq_ignore_ascii_case(s)) { return Ok(*value); } - let (high, low) = s - .split_once(':') - .ok_or_else(|| format!("community {s:?} must be ASN:VALUE or a well-known name"))?; - let parse = |part: &str| { - part.trim() - .parse::() - .map_err(|_| format!("community {s:?}: {part:?} is not a 16-bit number")) - }; + let (high, low) = s.split_once(':').ok_or_else(invalid)?; + let parse = |part: &str| part.trim().parse::().map_err(|_| invalid()); Ok((parse(high)?, parse(low)?)) } @@ -40,19 +46,14 @@ pub fn format_standard((high, low): (u16, u16)) -> String { format!("{high}:{low}") } -pub fn parse_large(s: &str) -> Result { +pub fn parse_large(s: &str) -> Result { let s = s.trim(); + let invalid = || CommunityError::Large(s.to_string()); let parts: Vec<&str> = s.split(':').collect(); if parts.len() != 3 { - return Err(format!( - "large community {s:?} must be GLOBAL:LOCAL1:LOCAL2" - )); + return Err(invalid()); } - let parse = |part: &str| { - part.trim() - .parse::() - .map_err(|_| format!("large community {s:?}: {part:?} is not a 32-bit number")) - }; + let parse = |part: &str| part.trim().parse::().map_err(|_| invalid()); Ok(LargeCommunity { global_admin: parse(parts[0])?, data_1: parse(parts[1])?, diff --git a/crates/packet/src/errors.rs b/crates/packet/src/errors.rs index aea255e..1412fc7 100644 --- a/crates/packet/src/errors.rs +++ b/crates/packet/src/errors.rs @@ -2,6 +2,27 @@ use serde::{Deserialize, Serialize}; use strum::FromRepr; use thiserror::Error; +/// A field holds a number that is not one of the values defined for it. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)] +#[error("invalid {what} value {value}")] +pub struct InvalidValue { + pub what: &'static str, + pub value: u32, +} + +/// A message that could not be decoded, together with the NOTIFICATION that +/// reports the problem to the peer (RFC 4271 §6). +#[derive(Debug, Clone, PartialEq, Eq, Error)] +#[error("{detail}")] +pub struct DecodeError { + /// The error code and subcode to send. + pub notification: BgpError, + /// The Data field of that NOTIFICATION, where the RFC defines one. + pub data: Vec, + /// What was wrong, for the log. + pub detail: String, +} + #[derive(Debug, Error)] #[error("Unknown BGP error code {code} subcode {subcode}")] pub struct UnknownErrorCode { @@ -102,6 +123,9 @@ impl TryFrom for MsgHeaderSubcode { #[derive(Debug, Clone, Copy, PartialEq, Eq, Error, Serialize, Deserialize, FromRepr)] #[repr(u8)] pub enum OpenMsgSubcode { + #[error("Unspecific")] + Unspecific = 0, + #[error("Unsupported BGP version number")] UnsupportedVersion = 1, diff --git a/crates/packet/src/message.rs b/crates/packet/src/message.rs index 57b2890..c646f68 100644 --- a/crates/packet/src/message.rs +++ b/crates/packet/src/message.rs @@ -4,10 +4,6 @@ use std::ops::Deref; use bitfield::bitfield; use bytes::BufMut; use bytes::BytesMut; -use eyre::Context; -use eyre::Result; -use eyre::bail; -use eyre::eyre; use nom::Err::Failure; use nom::IResult; use nom::Parser; @@ -22,9 +18,16 @@ use bgp_primitives::constants::AddressFamilyId; use bgp_primitives::constants::SubsequentAfi; use bgp_primitives::ip_prefix::IpPrefix; use bgp_primitives::parser::BgpParserError; +use bgp_primitives::parser::EncodeError; use bgp_primitives::parser::ParserContext; +use bgp_primitives::parser::ProtocolErrorValues; use crate::errors::BgpError; +use crate::errors::DecodeError; +use crate::errors::InvalidValue; +use crate::errors::MsgHeaderSubcode; +use crate::errors::OpenMsgSubcode; +use crate::errors::UpdateMsgSubcode; use crate::open::OpenMessage; /// A sentinel AS number used to denote that the actual AS number is provided as @@ -55,70 +58,124 @@ pub enum Message { } impl Message { + /// Decodes one complete BGP message, header included, from the start of + /// `buf`. Bytes after the message are ignored. + /// + /// A failure says which NOTIFICATION reports it, following RFC 4271 §6.1 + /// for the header and §6.2 and §6.3 for OPEN and UPDATE bodies. + pub fn decode(ctx: &ParserContext, buf: &[u8]) -> Result { + let header_error = |subcode, data: Vec, detail: String| DecodeError { + notification: BgpError::MsgHeader(subcode), + data, + detail, + }; + let header_len = usize::from(BGP_HEADER_LEN); + if buf.len() < header_len { + return Err(header_error( + MsgHeaderSubcode::BadMessageLength, + vec![], + format!("{} bytes is shorter than a BGP header", buf.len()), + )); + } + if buf[..16] != BGP_MARKER { + return Err(header_error( + MsgHeaderSubcode::ConnNotSynchronized, + vec![], + "invalid BGP marker: expected 16 x 0xFF".into(), + )); + } + let length = u16::from_be_bytes([buf[16], buf[17]]); + let msg_type = buf[18]; + if !(BGP_HEADER_LEN..=BGP_MAX_MSG_LEN).contains(&length) || usize::from(length) > buf.len() + { + return Err(header_error( + MsgHeaderSubcode::BadMessageLength, + length.to_be_bytes().to_vec(), + format!( + "BGP message length {length} is not between {BGP_HEADER_LEN} and \ + {BGP_MAX_MSG_LEN}, or exceeds the {} bytes received", + buf.len() + ), + )); + } + let body = &buf[header_len..usize::from(length)]; + let bad_length = |what: &str| { + header_error( + MsgHeaderSubcode::BadMessageLength, + length.to_be_bytes().to_vec(), + format!("{what} with a length of {length}"), + ) + }; + + match msg_type { + 1 => OpenMessage::from_wire(ctx, body) + .map(|(_, open)| Self::Open(open)) + .map_err(|e| { + let subcode = match &e { + nom::Err::Error(BgpParserError::ProtocolError(value)) + | nom::Err::Failure(BgpParserError::ProtocolError(value)) => match value { + ProtocolErrorValues::UnsupportedVersion(_) => { + OpenMsgSubcode::UnsupportedVersion + } + ProtocolErrorValues::UnknownOpenOption(_) => { + OpenMsgSubcode::UnsupportedOptionalParam + } + }, + _ => OpenMsgSubcode::Unspecific, + }; + // RFC 4271 §6.2: the Data field of Unsupported Version + // Number is the largest version we do support. + let data = match subcode { + OpenMsgSubcode::UnsupportedVersion => { + u16::from(BGP4_VERSION).to_be_bytes().to_vec() + } + _ => vec![], + }; + DecodeError { + notification: BgpError::OpenMsg(subcode), + data, + detail: format!("malformed OPEN: {}", describe(&e)), + } + }), + 2 => UpdateMessage::from_wire(ctx, body) + .map(|(_, update)| Self::Update(update)) + .map_err(|e| DecodeError { + notification: BgpError::UpdateMsg(UpdateMsgSubcode::MalformedAttrs), + data: vec![], + detail: format!("malformed UPDATE: {}", describe(&e)), + }), + 3 => NotificationMessage::from_wire(ctx, body) + .map(|(_, notification)| Self::Notification(notification)) + .map_err(|e| bad_length(&format!("malformed NOTIFICATION ({})", describe(&e)))), + 4 if body.is_empty() => Ok(Self::KeepAlive), + 4 => Err(bad_length("KEEPALIVE with a body")), + 5 => Ok(Self::RouteRefresh), + other => Err(header_error( + MsgHeaderSubcode::BadMessageType, + vec![other], + format!("unknown BGP message type {other}"), + )), + } + } + /// Parses a complete BGP message including the 19-byte header from `buf`. /// Returns the unparsed remainder and the decoded `Message`. + /// + /// This is [`Message::decode`] in the shape of the other parsers; use + /// `decode` when the reason for a failure matters. pub fn from_wire<'a>( ctx: &ParserContext, buf: &'a [u8], ) -> IResult<&'a [u8], Self, BgpParserError<&'a [u8]>> { - let (buf, marker) = nom::bytes::take(16usize).parse(buf)?; - if marker != BGP_MARKER { - return Err(Failure(BgpParserError::Eyre(eyre!( - "Invalid BGP marker: expected 16 x 0xFF" - )))); - } - - let (buf, length) = be_u16(buf)?; - if length < BGP_HEADER_LEN { - return Err(Failure(BgpParserError::Eyre(eyre!( - "BGP message length {length} is below minimum {BGP_HEADER_LEN}" - )))); - } - if length > BGP_MAX_MSG_LEN { - return Err(Failure(BgpParserError::Eyre(eyre!( - "BGP message length {length} exceeds maximum {BGP_MAX_MSG_LEN}" - )))); - } - - let (buf, msg_type) = be_u8(buf)?; - let body_len = length - BGP_HEADER_LEN; - let (buf, body) = nom::bytes::take(body_len as usize).parse(buf)?; - - let msg = match msg_type { - 1 => { - let (_, open) = OpenMessage::from_wire(ctx, body)?; - Self::Open(open) - } - 2 => { - let (_, update) = UpdateMessage::from_wire(ctx, body)?; - Self::Update(update) - } - 3 => { - let (_, notif) = NotificationMessage::from_wire(ctx, body)?; - Self::Notification(notif) - } - 4 => { - if !body.is_empty() { - return Err(Failure(BgpParserError::Eyre(eyre!( - "KEEPALIVE body must be empty, got {} bytes", - body.len() - )))); - } - Self::KeepAlive - } - 5 => Self::RouteRefresh, - t => { - return Err(Failure(BgpParserError::Eyre(eyre!( - "Unknown BGP message type {t}" - )))); - } - }; - - Ok((buf, msg)) + let msg = Self::decode(ctx, buf) + .map_err(|e| Failure(BgpParserError::Malformed(e.to_string())))?; + // `decode` checked that the length in the header is within `buf`. + let length = usize::from(u16::from_be_bytes([buf[16], buf[17]])); + Ok((&buf[length..], msg)) } /// Serializes the message (header + body) into `out`. - pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { // Serialize the body first so we can compute the total length. let mut body = BytesMut::new(); match self { @@ -128,10 +185,15 @@ impl Message { Self::KeepAlive | Self::RouteRefresh => {} } - let total_len = BGP_HEADER_LEN - .checked_add(body.len() as u16) + let total_len = usize::from(BGP_HEADER_LEN) + body.len(); + let total_len = u16::try_from(total_len) + .ok() .filter(|&l| l <= BGP_MAX_MSG_LEN) - .ok_or_else(|| eyre!("BGP message body too large to serialize"))?; + .ok_or(EncodeError::TooLong { + what: "BGP message", + len: total_len, + max: BGP_MAX_MSG_LEN.into(), + })?; out.put_slice(&BGP_MARKER); out.put_u16(total_len); @@ -140,7 +202,7 @@ impl Message { Ok(()) } - pub fn wire_len(&self, ctx: &ParserContext) -> Result { + pub fn wire_len(&self, ctx: &ParserContext) -> Result { let body_len: u16 = match self { Self::Open(m) => m.wire_len(ctx)?, Self::Update(m) => m.wire_len(ctx)?, @@ -187,7 +249,7 @@ impl UpdateMessage { nom::multi::many0(|buf| parse_prefix(AddressFamilyId::Ipv4, buf)) .parse(withdrawn_buf)?; if !withdrawn_rem.is_empty() { - return Err(Failure(BgpParserError::Eyre(eyre!( + return Err(Failure(BgpParserError::Malformed(format!( "Leftover bytes {:x?} after parsing withdrawn routes", withdrawn_rem )))); @@ -199,7 +261,7 @@ impl UpdateMessage { let (pa_rem, path_attributes) = nom::multi::many0(|buf| PathAttribute::from_wire(ctx, buf)).parse(pa_buf)?; if !pa_rem.is_empty() { - return Err(Failure(BgpParserError::Eyre(eyre!( + return Err(Failure(BgpParserError::Malformed(format!( "Leftover bytes {:x?} after parsing path attributes", pa_rem )))); @@ -211,7 +273,7 @@ impl UpdateMessage { let (buf, announced_routes) = nom::multi::many0(|buf| parse_prefix(AddressFamilyId::Ipv4, buf)).parse(buf)?; if !buf.is_empty() { - return Err(Failure(BgpParserError::Eyre(eyre!( + return Err(Failure(BgpParserError::Malformed(format!( "Leftover bytes {:x?} after parsing NLRI", buf )))); @@ -227,10 +289,9 @@ impl UpdateMessage { )) } - pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { // Withdrawn routes. - let withdrawn_len = prefix_list_wire_len(&self.withdrawn_routes) - .wrap_err("Withdrawn routes length exceeds u16::MAX")?; + let withdrawn_len = prefix_list_wire_len(&self.withdrawn_routes)?; out.put_u16(withdrawn_len); for prefix in &self.withdrawn_routes { out.put_u8(prefix.length); @@ -242,9 +303,9 @@ impl UpdateMessage { .path_attributes .iter() .map(|p| p.wire_len(ctx).map(|v| v as u32)) - .sum::>()?; - let pa_total = u16::try_from(pa_total) - .map_err(|_| eyre!("Total path attribute length exceeds u16::MAX"))?; + .sum::>()?; + let pa_total = + u16::try_from(pa_total).map_err(|_| too_long("path attributes", pa_total))?; out.put_u16(pa_total); for pa in &self.path_attributes { pa.to_wire(ctx, out)?; @@ -259,7 +320,7 @@ impl UpdateMessage { Ok(()) } - pub fn wire_len(&self, ctx: &ParserContext) -> Result { + pub fn wire_len(&self, ctx: &ParserContext) -> Result { // 2 bytes for the withdrawn length field + 2 bytes for the total path // attribute length field. let mut total: u32 = 4; @@ -272,16 +333,39 @@ impl UpdateMessage { for prefix in &self.announced_routes { total += 1 + prefix.prefix.len() as u32; } - u16::try_from(total).map_err(|_| eyre!("UPDATE message length exceeds u16::MAX")) + u16::try_from(total).map_err(|_| too_long("UPDATE message", total)) } } /// Computes the total wire length of a list of IPv4/IPv6 NLRI prefixes as they /// appear in the withdrawn or NLRI sections of an UPDATE message (each prefix /// is encoded as a 1-byte length followed by `ceil(length/8)` prefix bytes). -fn prefix_list_wire_len(prefixes: &[IpPrefix]) -> Result { +fn prefix_list_wire_len(prefixes: &[IpPrefix]) -> Result { let total: u32 = prefixes.iter().map(|p| 1 + p.prefix.len() as u32).sum(); - u16::try_from(total).map_err(|_| eyre!("Prefix list length exceeds u16::MAX")) + u16::try_from(total).map_err(|_| too_long("prefix list", total)) +} + +/// A parser failure as text for a log line. +fn describe(e: &nom::Err>) -> String { + match e { + nom::Err::Incomplete(_) => "truncated".into(), + nom::Err::Error(inner) | nom::Err::Failure(inner) => match inner { + BgpParserError::Malformed(text) => text.clone(), + BgpParserError::ProtocolError(value) => format!("{value:?}"), + BgpParserError::Nom(rest, kind) => { + format!("unexpected input ({kind:?}) with {} bytes left", rest.len()) + } + }, + } +} + +/// A length that does not fit the two-octet field that has to carry it. +fn too_long(what: &'static str, len: u32) -> EncodeError { + EncodeError::TooLong { + what, + len: len as usize, + max: u16::MAX.into(), + } } bitfield! { @@ -489,7 +573,7 @@ impl PathAttribute { Ok((buf, attr)) } - pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { macro_rules! write_path_attribute { ($out: expr, $attribute: expr) => { let wire_len = $attribute.wire_len(ctx)?; @@ -574,7 +658,7 @@ impl PathAttribute { /// Returns the total on-wire length of this path attribute, including the /// attribute header (flags + type + length) and the payload. - pub fn wire_len(&self, ctx: &ParserContext) -> Result { + pub fn wire_len(&self, ctx: &ParserContext) -> Result { let payload_len: u16 = match self { PathAttribute::Origin(a) => a.wire_len(ctx)?, PathAttribute::AsPath(a) => a.wire_len(ctx)?, @@ -592,8 +676,12 @@ impl PathAttribute { // An unknown attribute preserves its original extended_length // flag on re-serialization, so use it verbatim here rather than // re-deriving it from the payload length. - let payload_len = u16::try_from(payload.len()) - .map_err(|_| eyre!("UnknownPathAttribute payload exceeds u16::MAX"))?; + let payload_len = + u16::try_from(payload.len()).map_err(|_| EncodeError::TooLong { + what: "path attribute", + len: payload.len(), + max: u16::MAX.into(), + })?; let header_len: u16 = if flags.extended_length() { 4 } else { 3 }; return Ok(header_len + payload_len); } @@ -623,7 +711,7 @@ impl OriginPathAttribute { b if b == OriginPathAttribute::EGP as u8 => OriginPathAttribute::EGP, b if b == OriginPathAttribute::INCOMPLETE as u8 => OriginPathAttribute::INCOMPLETE, _ => { - return IResult::Err(nom::Err::Failure(BgpParserError::Eyre(eyre!( + return IResult::Err(nom::Err::Failure(BgpParserError::Malformed(format!( "Unknown Origin type {}", byte )))); @@ -632,25 +720,28 @@ impl OriginPathAttribute { Ok((buf, attr)) } - pub fn wire_len(&self, _ctx: &ParserContext) -> Result { + pub fn wire_len(&self, _ctx: &ParserContext) -> Result { Ok(1) } - pub fn to_wire(&self, _ctx: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, _ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { out.put_u8(*self as u8); Ok(()) } } impl TryFrom for OriginPathAttribute { - type Error = eyre::Error; + type Error = InvalidValue; - fn try_from(value: u8) -> Result { + fn try_from(value: u8) -> Result { match value { 0 => Ok(Self::IGP), 1 => Ok(Self::EGP), 2 => Ok(Self::INCOMPLETE), - other => bail!("Unexpected origin code {}", other), + other => Err(InvalidValue { + what: "ORIGIN", + value: other.into(), + }), } } } @@ -701,8 +792,8 @@ impl AsPathAttribute { .parse(buf) .map(|(buf, asns)| (buf, asns.iter().map(|asn| *asn as u32).collect()))?, None => { - return Err(nom::Err::Failure(BgpParserError::CustomText( - "Context must set four_octet_asn before being used", + return Err(nom::Err::Failure(BgpParserError::Malformed( + "Context must set four_octet_asn before being used".into(), ))); } }; @@ -722,9 +813,11 @@ impl AsPathAttribute { Ok((buf, Self { segments })) } - pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { if ctx.four_octet_asn.is_none_or(|val| !val) { - bail!("AsPathAttribute can only be sent for four_octet_asn enabled peers"); + return Err(EncodeError::Unencodable( + "AS_PATH can only be sent to peers with four-octet AS numbers", + )); } for segment in &self.segments { @@ -736,13 +829,16 @@ impl AsPathAttribute { .path .len() .try_into() - .wrap_err("AS Path length too long")?, + .map_err(|_| EncodeError::TooLong { + what: "AS path segment", + len: segment.path.len(), + max: u8::MAX.into(), + })?, ); // AS numbers. - match ctx - .four_octet_asn - .ok_or(eyre!("ctx.four_octet_asn must be set"))? - { + match ctx.four_octet_asn.ok_or(EncodeError::Unencodable( + "the AS number width has not been negotiated", + ))? { true => { for asn in &segment.path { out.put_u32(*asn); @@ -750,10 +846,9 @@ impl AsPathAttribute { } false => { for asn in &segment.path { - out.put_u16( - u16::try_from(*asn) - .map_err(|e| eyre!("AS number did not fit into u16: {}", e))?, - ); + out.put_u16(u16::try_from(*asn).map_err(|_| { + EncodeError::Unencodable("AS number does not fit in two octets") + })?); } } } @@ -762,13 +857,17 @@ impl AsPathAttribute { Ok(()) } - pub fn wire_len(&self, ctx: &ParserContext) -> Result { + pub fn wire_len(&self, ctx: &ParserContext) -> Result { let mut counter = 0; for segment in &self.segments { counter += match ctx.four_octet_asn { Some(true) => 2 + (4 * segment.path.len()), Some(false) => 2 + (2 * segment.path.len()), - None => bail!("ParserContext needs four_octet_asn set"), + None => { + return Err(EncodeError::Unencodable( + "the AS number width has not been negotiated", + )); + } }; } Ok(counter as u16) @@ -787,12 +886,12 @@ impl NextHopPathAttribute { Ok((buf, Self(Ipv4Addr::from(ip_u32)))) } - pub fn to_wire(&self, _ctx: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, _ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { out.put_u32(self.0.into()); Ok(()) } - pub fn wire_len(&self, _: &ParserContext) -> Result { + pub fn wire_len(&self, _: &ParserContext) -> Result { Ok(4) } } @@ -809,12 +908,12 @@ impl MultiExitDiscPathAttribute { Ok((buf, Self(val))) } - pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { out.put_u32(self.0); Ok(()) } - pub fn wire_len(&self, _: &ParserContext) -> Result { + pub fn wire_len(&self, _: &ParserContext) -> Result { Ok(4) } } @@ -831,12 +930,12 @@ impl LocalPrefPathAttribute { Ok((buf, Self(val))) } - pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { out.put_u32(self.0); Ok(()) } - pub fn wire_len(&self, _: &ParserContext) -> Result { + pub fn wire_len(&self, _: &ParserContext) -> Result { Ok(4) } } @@ -852,11 +951,11 @@ impl AtomicAggregatePathAttribute { Ok((buf, Self {})) } - pub fn to_wire(&self, _: &ParserContext, _: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, _: &ParserContext, _: &mut BytesMut) -> Result<(), EncodeError> { Ok(()) } - pub fn wire_len(&self, _: &ParserContext) -> Result { + pub fn wire_len(&self, _: &ParserContext) -> Result { Ok(0) } } @@ -873,8 +972,9 @@ impl AggregatorPathAttribute { buf: &'a [u8], ) -> IResult<&'a [u8], Self, BgpParserError<&'a [u8]>> { if ctx.four_octet_asn.is_none_or(|val| !val) { - return Err(nom::Err::Failure(BgpParserError::CustomText( - "AggregatorPathAttribute can only be parsed for four_octet_asn enabled peers", + return Err(nom::Err::Failure(BgpParserError::Malformed( + "AggregatorPathAttribute can only be parsed for four_octet_asn enabled peers" + .into(), ))); } let (buf, asn) = be_u32(buf)?; @@ -888,16 +988,18 @@ impl AggregatorPathAttribute { )) } - pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { if ctx.four_octet_asn.is_none_or(|val| !val) { - bail!("AggregatorPathAttribute can only be sent for four_octet_asn enabled peers"); + return Err(EncodeError::Unencodable( + "AGGREGATOR can only be sent to peers with four-octet AS numbers", + )); } out.put_u32(self.asn); out.put_u32(self.ip.into()); Ok(()) } - pub fn wire_len(&self, _: &ParserContext) -> Result { + pub fn wire_len(&self, _: &ParserContext) -> Result { Ok(8) } } @@ -914,7 +1016,7 @@ impl CommunitiesPathAttribute { Ok((buf, CommunitiesPathAttribute(values))) } - pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { for value in &self.0 { out.put_u16(value.0); out.put_u16(value.1); @@ -922,7 +1024,7 @@ impl CommunitiesPathAttribute { Ok(()) } - pub fn wire_len(&self, _: &ParserContext) -> Result { + pub fn wire_len(&self, _: &ParserContext) -> Result { Ok((self.0.len() * 4) as u16) } } @@ -948,18 +1050,18 @@ impl ExtendedCommunitiesPathAttribute { )) } - pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { for ec in &self.extended_communities { ec.to_wire(ctx, out)?; } Ok(()) } - pub fn wire_len(&self, ctx: &ParserContext) -> Result { + pub fn wire_len(&self, ctx: &ParserContext) -> Result { self.extended_communities .iter() .map(|ec| ec.wire_len(ctx)) - .sum::>() + .sum::>() } } @@ -1068,8 +1170,8 @@ impl ExtendedCommunity { _ => { let (buf, payload) = nom::bytes::take(5_usize).parse(buf)?; let value: [u8; 5] = payload.try_into().map_err(|_| { - Failure(BgpParserError::CustomText( - "Expected exactly 5 bytes from the parser", + Failure(BgpParserError::Malformed( + "Expected exactly 5 bytes from the parser".into(), )) })?; ( @@ -1086,7 +1188,7 @@ impl ExtendedCommunity { Ok((buf, parsed)) } - pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { match self { ExtendedCommunity::AsSpecific { typ, @@ -1145,7 +1247,7 @@ impl ExtendedCommunity { Ok(()) } - pub fn wire_len(&self, _: &ParserContext) -> Result { + pub fn wire_len(&self, _: &ParserContext) -> Result { Ok(8) } } @@ -1165,14 +1267,14 @@ impl LargeCommunitiesPathAttribute { Ok((buf, Self { communities })) } - pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { for community in &self.communities { community.to_wire(ctx, out)?; } Ok(()) } - pub fn wire_len(&self, _: &ParserContext) -> Result { + pub fn wire_len(&self, _: &ParserContext) -> Result { Ok(12_u16 * self.communities.len() as u16) } } @@ -1202,14 +1304,14 @@ impl LargeCommunity { )) } - pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { out.put_u32(self.global_admin); out.put_u32(self.data_1); out.put_u32(self.data_2); Ok(()) } - pub fn wire_len(&self, _: &ParserContext) -> Result { + pub fn wire_len(&self, _: &ParserContext) -> Result { Ok(12) } } @@ -1240,7 +1342,7 @@ impl std::fmt::Display for NlriNextHop { } impl NlriNextHop { - pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { match self { NlriNextHop::Ipv4(ipv4_addr) => out.put(&ipv4_addr.octets()[..]), NlriNextHop::Ipv6(ipv6_addr) => out.put(&ipv6_addr.octets()[..]), @@ -1271,7 +1373,7 @@ fn parse_prefix( let byte_len = prefix_len.div_ceil(8); let (buf, prefix_bytes) = nom::bytes::take(byte_len as usize).parse(buf)?; let prefix = IpPrefix::new(afi, prefix_bytes.to_vec(), prefix_len) - .map_err(|e| Failure(BgpParserError::Eyre(e)))?; + .map_err(|e| Failure(BgpParserError::Malformed(e.to_string())))?; Ok((buf, prefix)) } @@ -1290,12 +1392,12 @@ impl MpReachNlriPathAttribute { buf: &'a [u8], ) -> IResult<&'a [u8], Self, BgpParserError<&'a [u8]>> { let (buf, raw_afi) = be_u16(buf)?; - let afi = - AddressFamilyId::try_from(raw_afi).map_err(|e| Failure(BgpParserError::Eyre(e)))?; + let afi = AddressFamilyId::try_from(raw_afi) + .map_err(|e| Failure(BgpParserError::Malformed(e.to_string())))?; let (buf, raw_safi) = be_u8(buf)?; - let safi = - SubsequentAfi::try_from(raw_safi).map_err(|e| Failure(BgpParserError::Eyre(e)))?; + let safi = SubsequentAfi::try_from(raw_safi) + .map_err(|e| Failure(BgpParserError::Malformed(e.to_string())))?; let (buf, nh_len) = be_u8(buf)?; @@ -1303,7 +1405,7 @@ impl MpReachNlriPathAttribute { AddressFamilyId::Ipv4 => { // Read the length of the nexthop which should equal 4. if nh_len != 4 { - return Err(Failure(BgpParserError::Eyre(eyre!( + return Err(Failure(BgpParserError::Malformed(format!( "Got nexthop address length {} when expected 4 for IPv4 AFI", nh_len )))); @@ -1335,7 +1437,7 @@ impl MpReachNlriPathAttribute { NlriNextHop::Ipv6WithLl { global, link_local } } _ => { - return Err(Failure(BgpParserError::Eyre(eyre!( + return Err(Failure(BgpParserError::Malformed(format!( "Mismatched IPv6 nexthop length, got {}, want 16 or 32", nh_bytes.len() )))); @@ -1360,7 +1462,7 @@ impl MpReachNlriPathAttribute { )) } - pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { out.put_u16(self.afi as u16); out.put_u8(self.safi as u8); out.put_u8(self.next_hop.wire_len()); @@ -1373,7 +1475,7 @@ impl MpReachNlriPathAttribute { Ok(()) } - pub fn wire_len(&self, _: &ParserContext) -> Result { + pub fn wire_len(&self, _: &ParserContext) -> Result { Ok(4_u16 + self.next_hop.wire_len() as u16 + 1 // Reserved byte (SNPA). @@ -1398,12 +1500,12 @@ impl MpUnreachNlriPathAttribute { buf: &'a [u8], ) -> IResult<&'a [u8], Self, BgpParserError<&'a [u8]>> { let (buf, raw_afi) = be_u16(buf)?; - let afi = - AddressFamilyId::try_from(raw_afi).map_err(|e| Failure(BgpParserError::Eyre(e)))?; + let afi = AddressFamilyId::try_from(raw_afi) + .map_err(|e| Failure(BgpParserError::Malformed(e.to_string())))?; let (buf, raw_safi) = be_u8(buf)?; - let safi = - SubsequentAfi::try_from(raw_safi).map_err(|e| Failure(BgpParserError::Eyre(e)))?; + let safi = SubsequentAfi::try_from(raw_safi) + .map_err(|e| Failure(BgpParserError::Malformed(e.to_string())))?; let (buf, prefixes) = nom::multi::many0(|buf| parse_prefix(afi, buf)).parse(buf)?; @@ -1417,7 +1519,7 @@ impl MpUnreachNlriPathAttribute { )) } - pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { out.put_u16(self.afi as u16); out.put_u8(self.safi as u8); for prefix in &self.prefixes { @@ -1427,7 +1529,7 @@ impl MpUnreachNlriPathAttribute { Ok(()) } - pub fn wire_len(&self, _: &ParserContext) -> Result { + pub fn wire_len(&self, _: &ParserContext) -> Result { Ok(3 + self .prefixes .iter() @@ -1455,7 +1557,7 @@ impl NotificationMessage { let (buf, error_subcode) = be_u8(buf)?; let error = BgpError::from_code_subcode(error_code, error_subcode) - .map_err(|e| Failure(BgpParserError::Eyre(e.into())))?; + .map_err(|e| Failure(BgpParserError::Malformed(e.to_string())))?; Ok(( &[], @@ -1466,7 +1568,7 @@ impl NotificationMessage { )) } - pub fn to_wire(&self, _ctx: &ParserContext, buf: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, _ctx: &ParserContext, buf: &mut BytesMut) -> Result<(), EncodeError> { let (code, subcode) = self.error.code_subcode(); buf.put_u8(code); buf.put_u8(subcode); @@ -1996,4 +2098,93 @@ mod tests { buf[18] = 0x04; assert!(Message::from_wire(&ctx, &buf).is_err()); } + + /// A header, with `length` and `msg_type` as given, followed by `body`. + fn framed(length: u16, msg_type: u8, body: &[u8]) -> Vec { + let mut buf = vec![0xFF; 16]; + buf.extend_from_slice(&length.to_be_bytes()); + buf.push(msg_type); + buf.extend_from_slice(body); + buf + } + + /// RFC 4271 §6.1 to §6.3: what is wrong decides the NOTIFICATION, and + /// for a bad length or type the Data field carries the offending value. + #[test] + fn decode_failures_name_the_notification_to_send() { + use crate::errors::{ + BgpError, MsgHeaderSubcode as Header, OpenMsgSubcode, UpdateMsgSubcode, + }; + let ctx = default_v6_context(); + let failure = |buf: Vec| { + let e = Message::decode(&ctx, &buf).unwrap_err(); + (e.notification, e.data) + }; + + let mut bad_marker = framed(19, 4, &[]); + bad_marker[3] = 0; + assert_eq!( + failure(bad_marker), + (BgpError::MsgHeader(Header::ConnNotSynchronized), vec![]) + ); + // Shorter than a header, longer than the maximum, longer than the + // bytes at hand: all a bad length, reported with that length. + for length in [18u16, 4097, 100] { + assert_eq!( + failure(framed(length, 4, &[])), + ( + BgpError::MsgHeader(Header::BadMessageLength), + length.to_be_bytes().to_vec() + ), + "length {length}" + ); + } + assert_eq!( + failure(framed(20, 4, &[0])), + (BgpError::MsgHeader(Header::BadMessageLength), vec![0, 20]), + "a KEEPALIVE has no body" + ); + assert_eq!( + failure(framed(19, 9, &[])), + (BgpError::MsgHeader(Header::BadMessageType), vec![9]) + ); + // An OPEN for BGP version 3: the reply says we speak version 4. + assert_eq!( + failure(framed(29, 1, &[3, 0xfd, 0xe8, 0, 90, 10, 0, 0, 1, 0])), + ( + BgpError::OpenMsg(OpenMsgSubcode::UnsupportedVersion), + vec![0, 4] + ) + ); + // The same OPEN for version 4 is fine: no optional parameters is + // a valid OPEN. + let plain = framed(29, 1, &[4, 0xfd, 0xe8, 0, 90, 10, 0, 0, 1, 0]); + assert!( + matches!(Message::decode(&ctx, &plain), Ok(Message::Open(open)) if open.options.is_empty()) + ); + // An OPEN cut short. + assert_eq!( + failure(framed(21, 1, &[4, 0xfd])).0, + BgpError::OpenMsg(OpenMsgSubcode::Unspecific) + ); + // An UPDATE whose attribute length points past its end. + assert_eq!( + failure(framed(23, 2, &[0, 0, 0, 40])).0, + BgpError::UpdateMsg(UpdateMsgSubcode::MalformedAttrs) + ); + } + + #[test] + fn decode_ignores_bytes_after_the_message_and_from_wire_returns_them() { + let ctx = default_v6_context(); + let mut buf = framed(19, 4, &[]); + buf.extend_from_slice(&[1, 2, 3]); + + assert!(matches!( + Message::decode(&ctx, &buf), + Ok(Message::KeepAlive) + )); + let (rest, _) = Message::from_wire(&ctx, &buf).unwrap(); + assert_eq!(rest, &[1, 2, 3]); + } } diff --git a/crates/packet/src/open.rs b/crates/packet/src/open.rs index 1ba534d..7246940 100644 --- a/crates/packet/src/open.rs +++ b/crates/packet/src/open.rs @@ -2,24 +2,24 @@ use std::net::Ipv4Addr; use bitfield::bitfield; use bytes::{BufMut, BytesMut}; -use eyre::{Result, bail, eyre}; use nom::{ Err::Failure, IResult, Parser, - combinator::peek, number::complete::{be_u8, be_u16, be_u24, be_u32}, }; use serde::{Deserialize, Serialize}; + +use crate::errors::InvalidValue; use strum::EnumDiscriminants; use bgp_primitives::{ constants::{AddressFamilyId, SubsequentAfi}, - parser::{BgpParserError, ParserContext, ProtocolErrorValues}, + parser::{BgpParserError, EncodeError, ParserContext, ProtocolErrorValues}, }; macro_rules! fail { ( $($arg:expr),* $(,)? ) => { - Failure(BgpParserError::Eyre(eyre!($($arg),*))) + Failure(BgpParserError::Malformed(format!($($arg),*))) }; } @@ -50,7 +50,7 @@ impl OpenMessage { let (buf, version) = be_u8(buf)?; if version != 4 { return Err(Failure(BgpParserError::ProtocolError( - ProtocolErrorValues::UnknownOpenOption(version), + ProtocolErrorValues::UnsupportedVersion(version), ))); } @@ -62,10 +62,12 @@ impl OpenMessage { // To encode a capabilities length greater than 255 the packet structure // is prefixed with the extended length option (255), a length of 255, // and then a u16 containing the payload option. + // An OPEN without optional parameters ends at the length byte, so + // there may be nothing after it to look at. let (buf, opt_len) = be_u8(buf)?; - let (_, inner) = peek(be_u8).parse(buf)?; + let extended = opt_len == 255 && buf.first() == Some(&255); - let (buf, opt_len) = if inner == 255 { + let (buf, opt_len) = if extended { let (buf, _) = be_u8(buf)?; // Skip past the inner type. be_u16(buf)? } else { @@ -95,7 +97,7 @@ impl OpenMessage { )) } - pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { out.put_u8(self.version); out.put_u16(self.asn); out.put_u16(self.hold_time); @@ -107,7 +109,7 @@ impl OpenMessage { .options .iter() .map(|o| o.wire_len(ctx)) - .sum::>()?; + .sum::>()?; if opt_len > u8::MAX as u16 { out.put_u8(0xff); // Set length to 255 as per RFC 9072. out.put_u8(0xff); // Open option type for Ext length. @@ -123,13 +125,13 @@ impl OpenMessage { Ok(()) } - pub fn wire_len(&self, ctx: &ParserContext) -> Result { + pub fn wire_len(&self, ctx: &ParserContext) -> Result { let mut total = 10; let opt_len = self .options .iter() .map(|o| o.wire_len(ctx)) - .sum::>()?; + .sum::>()?; if opt_len > u8::MAX as u16 { total += opt_len + 3; // 1+2 bytes extra for the type + extended length. } else { @@ -177,12 +179,19 @@ impl OpenOption { } } - pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { match self { OpenOption::Capabilities(items) => { - let cap_len = items.iter().map(|i| i.wire_len(ctx)).sum::>()?; + let cap_len = items + .iter() + .map(|i| i.wire_len(ctx)) + .sum::>()?; if cap_len > u8::MAX as u16 { - bail!("Cannot write Capabilities with more than u8::MAX length"); + return Err(EncodeError::TooLong { + what: "capabilities", + len: cap_len.into(), + max: u8::MAX.into(), + }); } out.put_u8(OpenOptionDiscriminants::Capabilities as u8); out.put_u8(cap_len as u8); @@ -194,12 +203,12 @@ impl OpenOption { Ok(()) } - pub fn wire_len(&self, ctx: &ParserContext) -> Result { + pub fn wire_len(&self, ctx: &ParserContext) -> Result { match self { OpenOption::Capabilities(items) => items .iter() .map(|i| i.wire_len(ctx)) - .sum::>() + .sum::>() .map(|v| v + 2), } } @@ -329,10 +338,9 @@ impl Capability { macro_rules! require_empty { ($buf:expr, $cap_type:expr) => { if !$buf.is_empty() { - return Err(nom::Err::Failure(BgpParserError::Eyre(eyre!( + return Err(nom::Err::Failure(BgpParserError::Malformed(format!( "Unexpected leftover bytes {:x?} while parsing {} capability", - $buf, - $cap_type + $buf, $cap_type )))); } }; @@ -548,7 +556,7 @@ impl Capability { } } - pub fn to_wire(&self, _ctx: &ParserContext, out: &mut BytesMut) -> Result<()> { + pub fn to_wire(&self, _ctx: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { out.put_u8(self.discriminant()); out.put_u8(self.payload_len()); match self { @@ -599,7 +607,11 @@ impl Capability { flags_restart_time |= 1 << 15; } if *restart_time > 0xfff { - bail!("Restart time too large to be encoded: {}", restart_time); + return Err(EncodeError::TooLong { + what: "graceful restart time", + len: (*restart_time).into(), + max: 0xfff, + }); } flags_restart_time |= restart_time; out.put_u16(flags_restart_time); @@ -639,7 +651,7 @@ impl Capability { Ok(()) } - pub fn wire_len(&self, _ctx: &ParserContext) -> Result { + pub fn wire_len(&self, _ctx: &ParserContext) -> Result { Ok(2 + self.payload_len() as u16) } @@ -718,15 +730,18 @@ pub enum OrfSendRecv { } impl TryFrom for OrfSendRecv { - type Error = eyre::ErrReport; + type Error = InvalidValue; - fn try_from(value: u8) -> Result { - Ok(match value { - value if value == OrfSendRecv::Receive as u8 => OrfSendRecv::Receive, - value if value == OrfSendRecv::Send as u8 => OrfSendRecv::Send, - value if value == OrfSendRecv::Both as u8 => OrfSendRecv::Both, - other => bail!("Invalid ORF send/recv value: {}", other), - }) + fn try_from(value: u8) -> Result { + match value { + value if value == OrfSendRecv::Receive as u8 => Ok(OrfSendRecv::Receive), + value if value == OrfSendRecv::Send as u8 => Ok(OrfSendRecv::Send), + value if value == OrfSendRecv::Both as u8 => Ok(OrfSendRecv::Both), + other => Err(InvalidValue { + what: "ORF send/receive", + value: other.into(), + }), + } } } @@ -739,15 +754,18 @@ pub enum AddPathSendRecv { } impl TryFrom for AddPathSendRecv { - type Error = eyre::ErrReport; + type Error = InvalidValue; - fn try_from(value: u8) -> Result { - Ok(match value { - value if value == AddPathSendRecv::Receive as u8 => AddPathSendRecv::Receive, - value if value == AddPathSendRecv::Send as u8 => AddPathSendRecv::Send, - value if value == AddPathSendRecv::Both as u8 => AddPathSendRecv::Both, - other => bail!("Invalid AddPathSendRecv send/recv value: {}", other), - }) + fn try_from(value: u8) -> Result { + match value { + value if value == AddPathSendRecv::Receive as u8 => Ok(AddPathSendRecv::Receive), + value if value == AddPathSendRecv::Send as u8 => Ok(AddPathSendRecv::Send), + value if value == AddPathSendRecv::Both as u8 => Ok(AddPathSendRecv::Both), + other => Err(InvalidValue { + what: "ADD-PATH send/receive", + value: other.into(), + }), + } } } diff --git a/crates/primitives/Cargo.toml b/crates/primitives/Cargo.toml index df5ca5d..75c555e 100644 --- a/crates/primitives/Cargo.toml +++ b/crates/primitives/Cargo.toml @@ -11,7 +11,6 @@ version.workspace = true [dependencies] bitfield.workspace = true bytes.workspace = true -eyre.workspace = true nom.workspace = true serde.workspace = true serde_repr.workspace = true diff --git a/crates/primitives/src/constants.rs b/crates/primitives/src/constants.rs index d9bdf4d..704250f 100644 --- a/crates/primitives/src/constants.rs +++ b/crates/primitives/src/constants.rs @@ -1,7 +1,17 @@ use std::fmt::Display; -use eyre::bail; use serde_repr::{Deserialize_repr, Serialize_repr}; +use thiserror::Error; + +/// An address family number this implementation has no name for. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)] +#[error("unknown address family identifier {0}")] +pub struct UnknownAfi(pub u16); + +/// A subsequent address family number this implementation has no name for. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)] +#[error("unknown subsequent address family identifier {0}")] +pub struct UnknownSafi(pub u16); #[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, Serialize_repr, Deserialize_repr)] #[repr(u16)] @@ -11,14 +21,14 @@ pub enum AddressFamilyId { } impl TryFrom for AddressFamilyId { - type Error = eyre::Error; + type Error = UnknownAfi; fn try_from(value: u16) -> Result { - Ok(match value { - x if x == Self::Ipv4 as u16 => Self::Ipv4, - x if x == Self::Ipv6 as u16 => Self::Ipv6, - other => bail!("Unknown AddressFamily: {}", other), - }) + match value { + x if x == Self::Ipv4 as u16 => Ok(Self::Ipv4), + x if x == Self::Ipv6 as u16 => Ok(Self::Ipv6), + other => Err(UnknownAfi(other)), + } } } @@ -43,27 +53,28 @@ pub enum SubsequentAfi { } impl TryFrom for SubsequentAfi { - type Error = eyre::Error; + type Error = UnknownSafi; fn try_from(value: u8) -> Result { - Ok(match value { - x if x == Self::Unicast as u8 => Self::Unicast, - x if x == Self::Multicast as u8 => Self::Multicast, - x if x == Self::NlriWithMpls as u8 => Self::NlriWithMpls, - x if x == Self::MplsLabelledVpn as u8 => Self::MplsLabelledVpn, - x if x == Self::MulticastMplsVpn as u8 => Self::MulticastMplsVpn, - other => bail!("Unknown SubsequentAfi: {}", other), - }) + match value { + x if x == Self::Unicast as u8 => Ok(Self::Unicast), + x if x == Self::Multicast as u8 => Ok(Self::Multicast), + x if x == Self::NlriWithMpls as u8 => Ok(Self::NlriWithMpls), + x if x == Self::MplsLabelledVpn as u8 => Ok(Self::MplsLabelledVpn), + x if x == Self::MulticastMplsVpn as u8 => Ok(Self::MulticastMplsVpn), + other => Err(UnknownSafi(other.into())), + } } } -/// The implementation try_from for u16 just reuses the u8 implementation since -/// we don't have any cases where the SAFI is longer than u8::MAX yet. +/// Some encodings carry the SAFI in two octets. Every SAFI we know fits in +/// one, so a larger value is unknown, not a smaller one in disguise. impl TryFrom for SubsequentAfi { - type Error = eyre::Error; + type Error = UnknownSafi; fn try_from(value: u16) -> Result { - TryFrom::::try_from(value as u8) + let narrow = u8::try_from(value).map_err(|_| UnknownSafi(value))?; + Self::try_from(narrow) } } @@ -78,3 +89,16 @@ impl Display for SubsequentAfi { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_wide_safi_is_not_truncated_into_a_known_one() { + assert_eq!(SubsequentAfi::try_from(1u16), Ok(SubsequentAfi::Unicast)); + // 257 has the low byte of Unicast. + assert_eq!(SubsequentAfi::try_from(257u16), Err(UnknownSafi(257))); + assert_eq!(AddressFamilyId::try_from(3u16), Err(UnknownAfi(3))); + } +} diff --git a/crates/primitives/src/ip_prefix.rs b/crates/primitives/src/ip_prefix.rs index 6b28a79..933fe7e 100644 --- a/crates/primitives/src/ip_prefix.rs +++ b/crates/primitives/src/ip_prefix.rs @@ -3,11 +3,28 @@ use std::net::{Ipv4Addr, Ipv6Addr}; use std::str::FromStr; use bytes::{Buf, BufMut, BytesMut}; -use eyre::{Result, bail}; use serde::{Deserialize, Serialize}; +use thiserror::Error; use crate::constants::AddressFamilyId; -use crate::parser::{ParserContext, ToWireError}; +use crate::parser::{EncodeError, ParserContext}; + +/// Why something is not a valid prefix. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum PrefixError { + #[error("prefix {0:?} is not in address/length form")] + Format(String), + #[error("prefix {0:?} has an address that is neither IPv4 nor IPv6")] + Address(String), + #[error("prefix length {length} is longer than the {max} bits of an address")] + LengthTooLong { length: u8, max: u8 }, + #[error("prefix length {length} needs {needed} address bytes, got {got}")] + TooFewBytes { + length: u8, + needed: usize, + got: usize, + }, +} /// IpPrefix represents some IP address prefix, for a specific AddressFamilyId. #[derive(Debug, PartialEq, Eq, Clone, Hash, Serialize, Deserialize)] @@ -18,14 +35,19 @@ pub struct IpPrefix { } impl IpPrefix { - pub fn new(address_family: AddressFamilyId, prefix: Vec, length: u8) -> Result { + pub fn new( + address_family: AddressFamilyId, + prefix: Vec, + length: u8, + ) -> Result { // Ensure that the prefix we are given contains the right number of bytes corresponding to the prefix length. - if prefix.len() < length.div_ceil(8).into() { - bail!( - "Mismatched prefix {:?} for given prefix length: {}", - prefix, - length - ); + let needed = usize::from(length.div_ceil(8)); + if prefix.len() < needed { + return Err(PrefixError::TooFewBytes { + length, + needed, + got: prefix.len(), + }); } Ok(Self { address_family, @@ -53,10 +75,10 @@ impl IpPrefix { (self.prefix[full_bytes] & mask) == (other.prefix[full_bytes] & mask) } - pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<(), ToWireError> { + pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<(), EncodeError> { // Verify that there is enough space to write the IpPrefix. if out.remaining() < (self.prefix.len() + 1) { - Err(ToWireError::OutBufferOverflow)?; + return Err(EncodeError::OutBufferOverflow); } // Write length and prefix. @@ -68,32 +90,23 @@ impl IpPrefix { } impl TryFrom<&str> for IpPrefix { - type Error = eyre::Error; + type Error = PrefixError; fn try_from(value: &str) -> Result { - let parts: Vec<&str> = value.split("/").collect(); + let format = || PrefixError::Format(value.to_string()); + let address = || PrefixError::Address(value.to_string()); + let (addr, length) = value.split_once('/').ok_or_else(format)?; + let length: u8 = length.parse().map_err(|_| format())?; - if parts.len() != 2 { - bail!( - "Expected IpPrefix in format prefix/length but got: {}", - value - ); - } - - let length: u8 = parts[1].parse().map_err(eyre::Error::from)?; - let mut octets; - let afi: AddressFamilyId; - - if parts[0].contains(":") { - afi = AddressFamilyId::Ipv6; - let addr: Ipv6Addr = Ipv6Addr::from_str(parts[0]).map_err(eyre::Error::from)?; - octets = addr.octets().to_vec(); - } else if parts[0].contains(".") { - afi = AddressFamilyId::Ipv4; - let addr: Ipv4Addr = Ipv4Addr::from_str(parts[0]).map_err(eyre::Error::from)?; - octets = addr.octets().to_vec(); + let (afi, mut octets, max) = if addr.contains(':') { + let addr = Ipv6Addr::from_str(addr).map_err(|_| address())?; + (AddressFamilyId::Ipv6, addr.octets().to_vec(), 128) } else { - bail!("Could not figure out address type") + let addr = Ipv4Addr::from_str(addr).map_err(|_| address())?; + (AddressFamilyId::Ipv4, addr.octets().to_vec(), 32) + }; + if length > max { + return Err(PrefixError::LengthTooLong { length, max }); } // Truncate the octets we have to the right number of bytes to match the prefix length.. @@ -154,7 +167,7 @@ impl Display for IpPrefix { mod tests { use crate::constants::AddressFamilyId; - use super::IpPrefix; + use super::{IpPrefix, PrefixError}; macro_rules! verify_roundtrip { ($name:ident, $prefix_str:expr, $afi:expr, $bytes:expr, $length:expr) => { @@ -224,6 +237,42 @@ mod tests { 32 ); + #[test] + fn text_that_is_not_a_prefix_is_rejected_with_the_reason() { + for (text, expected) in [ + ("10.0.0.0", PrefixError::Format("10.0.0.0".into())), + ("10.0.0.0/x", PrefixError::Format("10.0.0.0/x".into())), + ("10.0.0/8", PrefixError::Address("10.0.0/8".into())), + ("nonsense/8", PrefixError::Address("nonsense/8".into())), + ( + "10.0.0.0/33", + PrefixError::LengthTooLong { + length: 33, + max: 32, + }, + ), + ( + "2001:db8::/129", + PrefixError::LengthTooLong { + length: 129, + max: 128, + }, + ), + ] { + assert_eq!(IpPrefix::try_from(text), Err(expected), "{text}"); + } + assert!(IpPrefix::try_from("0.0.0.0/0").is_ok()); + assert!(IpPrefix::try_from("2001:db8::1/128").is_ok()); + assert_eq!( + IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 16), + Err(PrefixError::TooFewBytes { + length: 16, + needed: 2, + got: 1 + }) + ); + } + #[test] fn contains_covers_more_specifics() { let p = |s: &str| IpPrefix::try_from(s).unwrap(); diff --git a/crates/primitives/src/parser.rs b/crates/primitives/src/parser.rs index d1cca4c..12d614a 100644 --- a/crates/primitives/src/parser.rs +++ b/crates/primitives/src/parser.rs @@ -1,6 +1,5 @@ -use std::fmt::Display; - use nom::error::{ErrorKind, ParseError}; +use thiserror::Error; use crate::constants::AddressFamilyId; @@ -12,33 +11,33 @@ pub struct ParserContext { pub address_family: Option, } -#[derive(Debug)] -pub enum ToWireError { - /// There was not enough space in the output buffer to serialize the data into. +/// Why a value could not be written to the wire. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum EncodeError { + /// There was not enough space in the output buffer. + #[error("output buffer too small")] OutBufferOverflow, - /// Another error. - Other(eyre::Error), + /// A length does not fit the field that has to carry it. + #[error("{what} is too long to encode: {len} exceeds {max}")] + TooLong { + what: &'static str, + len: usize, + max: usize, + }, + /// The value has no encoding in this context, e.g. a four-octet AS + /// number on a session that negotiated two-octet ones. + #[error("{0}")] + Unencodable(&'static str), } -impl Display for ToWireError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - ToWireError::OutBufferOverflow => write!(f, "OutBufferOverflow"), - ToWireError::Other(report) => report.fmt(f), - } - } -} - -impl std::error::Error for ToWireError {} - // Custom error type for the parser. #[derive(Debug)] pub enum BgpParserError { /// ProtocolError encapsulates protocol level semantics problems that require /// action from the caller of the parser. ProtocolError(ProtocolErrorValues), - CustomText(&'static str), - Eyre(eyre::ErrReport), + /// The input does not follow the format; the text says how. + Malformed(String), Nom(I, ErrorKind), } diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 4892562..b6f3823 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -14,7 +14,6 @@ bgp-rib.workspace = true bytes.workspace = true clap.workspace = true futures.workspace = true -eyre.workspace = true prometheus.workspace = true tokio.workspace = true tracing-subscriber.workspace = true diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index ceb145c..50be151 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -6,9 +6,9 @@ use bgp_packet::community; use bgp_packet::message::OriginPathAttribute; use bgp_primitives::ip_prefix::IpPrefix; use bgp_rib::{LocalRoute, LocalRouteAttrs}; -use eyre::{Result, WrapErr, bail, eyre}; -use crate::proto::config::{BgpConfig, OriginatedRoute}; +use crate::error::{ConfigError, OriginateProblem, PeerProblem}; +use crate::proto::config::{BgpConfig, OriginatedRoute, Peer}; const DEFAULT_BGP_PORT: u16 = 179; const DEFAULT_HOLD_TIME_SECS: u16 = 90; @@ -48,16 +48,12 @@ pub struct PeerConfig { } impl Config { - pub fn from_proto(proto: BgpConfig) -> Result { - let router_id = Ipv4Addr::from_str(&proto.router_id).map_err(|_| { - eyre!( - "Invalid router_id {:?}: must be a dotted-quad IPv4 address", - proto.router_id - ) - })?; + pub fn from_proto(proto: BgpConfig) -> Result { + let router_id = Ipv4Addr::from_str(&proto.router_id) + .map_err(|_| ConfigError::RouterId(proto.router_id.clone()))?; if proto.local_asn == 0 { - bail!("local_asn must be non-zero"); + return Err(ConfigError::LocalAsn); } let listen_endpoints = if proto.listen_endpoints.is_empty() { @@ -71,72 +67,42 @@ impl Config { } else { proto .listen_endpoints - .into_iter() + .iter() .map(|ep| parse_endpoint(&ep.address, ep.port)) - .collect::>()? + .collect::>()? }; - let peers = proto.peers + let peers = proto + .peers .into_iter() - .map(|p| { - let ep = p.endpoint.as_ref() - .ok_or_else(|| eyre!("Peer {:?} is missing endpoint", p.name))?; - let endpoint = parse_endpoint(&ep.address, ep.port)?; - - if p.remote_asn == 0 { - bail!("Peer {:?} remote_asn must be non-zero", p.name); - } - - let hold_time_secs = if p.hold_time_seconds == 0 { - DEFAULT_HOLD_TIME_SECS - } else if p.hold_time_seconds < 3 { - bail!("Peer {:?} hold_time_seconds must be 0 or >= 3", p.name); - } else { - p.hold_time_seconds as u16 - }; - - let local_address = if p.local_address.is_empty() { - None - } else { - let ip = IpAddr::from_str(&p.local_address) - .map_err(|_| eyre!("Peer {:?} has invalid local_address {:?}", p.name, p.local_address))?; - if ip.is_ipv4() != endpoint.is_ipv4() { - bail!("Peer {:?}: local_address {ip} and endpoint {endpoint} are different address families", p.name); - } - Some(ip) - }; - - Ok(PeerConfig { - name: p.name, - endpoint, - remote_asn: p.remote_asn, - hold_time_secs, - local_address, + .map(|peer| { + let name = peer.name.clone(); + parse_peer(peer).map_err(|problem| ConfigError::Peer { + peer: name, + problem, }) }) - .collect::>()?; + .collect::>()?; - let originate: Vec = proto - .originate - .iter() - .map(parse_originated_route) - .collect::>()?; - for (i, route) in originate.iter().enumerate() { - if originate[..i] + let mut originate: Vec = Vec::with_capacity(proto.originate.len()); + for entry in &proto.originate { + let rejected = |problem| ConfigError::Originate { + prefix: entry.prefix.clone(), + problem, + }; + let route = parse_originated_route(entry).map_err(rejected)?; + if originate .iter() .any(|earlier| earlier.prefix == route.prefix) { - bail!("originate lists {} more than once", route.prefix); + return Err(rejected(OriginateProblem::Duplicate)); } + originate.push(route); } let http_endpoint = match proto.http_endpoint.into_option() { - Some(ep) => { - if ep.port == 0 { - bail!("http_endpoint.port is required"); - } - Some(parse_endpoint(&ep.address, ep.port).wrap_err("Invalid http_endpoint")?) - } + Some(ep) if ep.port == 0 => return Err(ConfigError::HttpPort), + Some(ep) => Some(parse_endpoint(&ep.address, ep.port)?), None => None, }; @@ -160,49 +126,80 @@ impl Config { } } -fn parse_endpoint(address: &str, port: u32) -> Result { - let port = if port == 0 { - DEFAULT_BGP_PORT - } else if port > u16::MAX as u32 { - bail!("Port {port} is out of range"); - } else { - port as u16 +fn parse_peer(peer: Peer) -> Result { + let ep = peer.endpoint.as_ref().ok_or(PeerProblem::MissingEndpoint)?; + let endpoint = + parse_endpoint(&ep.address, ep.port).map_err(|e| PeerProblem::Endpoint(Box::new(e)))?; + + if peer.remote_asn == 0 { + return Err(PeerProblem::RemoteAsn); + } + + let hold_time_secs = match peer.hold_time_seconds { + 0 => DEFAULT_HOLD_TIME_SECS, + // RFC 4271 §4.2: zero, or at least three seconds. + 1 | 2 => return Err(PeerProblem::HoldTime), + secs => u16::try_from(secs).map_err(|_| PeerProblem::HoldTime)?, }; - let ip = IpAddr::from_str(address).map_err(|_| eyre!("Invalid IP address {address:?}"))?; + let local_address = if peer.local_address.is_empty() { + None + } else { + let ip = IpAddr::from_str(&peer.local_address) + .map_err(|_| PeerProblem::LocalAddress(peer.local_address.clone()))?; + if ip.is_ipv4() != endpoint.is_ipv4() { + return Err(PeerProblem::AddressFamilyMismatch); + } + Some(ip) + }; + Ok(PeerConfig { + name: peer.name, + endpoint, + remote_asn: peer.remote_asn, + hold_time_secs, + local_address, + }) +} + +fn parse_endpoint(address: &str, port: u32) -> Result { + let port = match port { + 0 => DEFAULT_BGP_PORT, + port => u16::try_from(port).map_err(|_| ConfigError::Port(port))?, + }; + let ip = IpAddr::from_str(address).map_err(|_| ConfigError::Address(address.to_string()))?; Ok(SocketAddr::new(ip, port)) } -fn parse_originated_route(route: &OriginatedRoute) -> Result { - let prefix = IpPrefix::try_from(route.prefix.as_str()) - .wrap_err_with(|| format!("Invalid originate prefix {:?}", route.prefix))?; +fn parse_originated_route(route: &OriginatedRoute) -> Result { + let prefix = IpPrefix::try_from(route.prefix.as_str())?; let origin = match route.origin.to_ascii_uppercase().as_str() { "" | "IGP" => OriginPathAttribute::IGP, "EGP" => OriginPathAttribute::EGP, "INCOMPLETE" => OriginPathAttribute::INCOMPLETE, - other => bail!("originate {prefix}: origin {other:?} must be IGP, EGP or INCOMPLETE"), + _ => return Err(OriginateProblem::Origin(route.origin.clone())), }; - if route.as_path_prepend > MAX_AS_PATH_PREPEND { - bail!("originate {prefix}: as_path_prepend must be at most {MAX_AS_PATH_PREPEND}"); - } + let prepend = u8::try_from(route.as_path_prepend) + .ok() + .filter(|&n| u32::from(n) <= MAX_AS_PATH_PREPEND) + .ok_or(OriginateProblem::Prepend { + max: MAX_AS_PATH_PREPEND, + })?; let communities = route .communities .iter() .map(|c| community::parse_standard(c)) - .collect::>() - .map_err(|e| eyre!("originate {prefix}: {e}"))?; + .collect::>()?; let large_communities = route .large_communities .iter() .map(|c| community::parse_large(c)) - .collect::>() - .map_err(|e| eyre!("originate {prefix}: {e}"))?; + .collect::>()?; Ok(LocalRoute::new( prefix, LocalRouteAttrs { origin, - prepend: route.as_path_prepend as u8, + prepend, med: route.med, communities, large_communities, @@ -212,14 +209,20 @@ fn parse_originated_route(route: &OriginatedRoute) -> Result { #[cfg(test)] mod tests { + use bgp_primitives::ip_prefix::PrefixError; + use super::*; - fn parse(text: &str) -> Result { - let proto = - protobuf::text_format::parse_from_str::(text).map_err(|e| eyre!("{e}"))?; + fn parse(text: &str) -> Result { + let proto = protobuf::text_format::parse_from_str::(text) + .expect("the test's textproto is well formed"); Config::from_proto(proto) } + fn rejected(text: &str) -> ConfigError { + parse(text).err().expect(text) + } + const BASE: &str = "router_id: \"192.0.2.1\" local_asn: 65000\n"; #[test] @@ -252,33 +255,90 @@ mod tests { } #[test] - fn bad_originated_routes_are_rejected_with_the_prefix_named() { - for (body, expected) in [ - ("prefix: \"not-a-prefix\"", "Invalid originate prefix"), - ( - "prefix: \"192.0.2.0/24\" origin: \"static\"", - "must be IGP, EGP or INCOMPLETE", - ), - ("prefix: \"192.0.2.0/24\" as_path_prepend: 17", "at most 16"), - ( - "prefix: \"192.0.2.0/24\" communities: \"70000:1\"", - "originate 192.0.2.0/24", - ), - ( - "prefix: \"192.0.2.0/24\" large_communities: \"1:2\"", - "originate 192.0.2.0/24", - ), - ] { - let err = parse(&format!("{BASE}originate {{ {body} }}")) - .err() - .expect(body); - assert!(format!("{err:#}").contains(expected), "{body}: {err:#}"); - } - let err = parse(&format!( + fn bad_originated_routes_are_rejected_with_the_entry_and_the_reason() { + let problem = |body: &str| match rejected(&format!("{BASE}originate {{ {body} }}")) { + ConfigError::Originate { problem, .. } => problem, + other => panic!("{body}: {other}"), + }; + + assert_eq!( + problem("prefix: \"not-a-prefix\""), + OriginateProblem::Prefix(PrefixError::Format("not-a-prefix".into())) + ); + assert_eq!( + problem("prefix: \"192.0.2.0/24\" origin: \"static\""), + OriginateProblem::Origin("static".into()) + ); + assert_eq!( + problem("prefix: \"192.0.2.0/24\" as_path_prepend: 17"), + OriginateProblem::Prepend { max: 16 } + ); + assert!(matches!( + problem("prefix: \"192.0.2.0/24\" communities: \"70000:1\""), + OriginateProblem::Community(_) + )); + assert!(matches!( + problem("prefix: \"192.0.2.0/24\" large_communities: \"1:2\""), + OriginateProblem::Community(_) + )); + + let duplicate = rejected(&format!( "{BASE}originate {{ prefix: \"192.0.2.0/24\" }} originate {{ prefix: \"192.0.2.0/24\" med: 5 }}" - )) - .err() - .unwrap(); - assert!(format!("{err:#}").contains("more than once"), "{err:#}"); + )); + assert_eq!( + duplicate, + ConfigError::Originate { + prefix: "192.0.2.0/24".into(), + problem: OriginateProblem::Duplicate + } + ); + // The message names the entry, so the operator can find it. + assert_eq!( + duplicate.to_string(), + "originate \"192.0.2.0/24\": listed more than once" + ); + } + + #[test] + fn bad_peers_and_endpoints_are_rejected_with_the_peer_named() { + assert_eq!( + rejected("router_id: \"nope\" local_asn: 1"), + ConfigError::RouterId("nope".into()) + ); + assert_eq!(rejected("router_id: \"192.0.2.1\""), ConfigError::LocalAsn); + + let problem = |body: &str| match rejected(&format!("{BASE}peers {{ name: \"r1\" {body} }}")) + { + ConfigError::Peer { peer, problem } => { + assert_eq!(peer, "r1"); + problem + } + other => panic!("{body}: {other}"), + }; + let endpoint = "endpoint { address: \"192.0.2.2\" }"; + assert_eq!(problem("remote_asn: 1"), PeerProblem::MissingEndpoint); + assert_eq!(problem(endpoint), PeerProblem::RemoteAsn); + assert_eq!( + problem(&format!("{endpoint} remote_asn: 1 hold_time_seconds: 2")), + PeerProblem::HoldTime + ); + assert_eq!( + problem(&format!( + "{endpoint} remote_asn: 1 hold_time_seconds: 70000" + )), + PeerProblem::HoldTime + ); + assert_eq!( + problem(&format!("{endpoint} remote_asn: 1 local_address: \"::1\"")), + PeerProblem::AddressFamilyMismatch + ); + assert_eq!( + problem("endpoint { address: \"192.0.2.2\" port: 70000 } remote_asn: 1"), + PeerProblem::Endpoint(Box::new(ConfigError::Port(70000))) + ); + assert_eq!( + rejected(&format!("{BASE}http_endpoint {{ address: \"127.0.0.1\" }}")), + ConfigError::HttpPort + ); } } diff --git a/crates/server/src/dialer.rs b/crates/server/src/dialer.rs index 151f435..376648f 100644 --- a/crates/server/src/dialer.rs +++ b/crates/server/src/dialer.rs @@ -87,16 +87,17 @@ async fn dial_loop(peer: Arc, config: SessionConfig, connect_retry: run_session(stream, peer.endpoint, config.clone()) .await .map(|()| "session closed by peer".to_string()) + .map_err(|e| e.to_string()) } - Ok(Err(e)) => Err(eyre::eyre!("connect failed: {e}")), - Err(_) => Err(eyre::eyre!("connect timed out after {CONNECT_TIMEOUT:?}")), + Ok(Err(e)) => Err(format!("connect failed: {e}")), + Err(_) => Err(format!("connect timed out after {CONNECT_TIMEOUT:?}")), }; let reason = match outcome { Ok(msg) => msg, - Err(e) => { - warn!(peer = %peer.name, endpoint = %peer.endpoint, err = %e, "Session attempt ended"); - e.to_string() + Err(reason) => { + warn!(peer = %peer.name, endpoint = %peer.endpoint, err = %reason, "Session attempt ended"); + reason } }; diff --git a/crates/server/src/error.rs b/crates/server/src/error.rs new file mode 100644 index 0000000..d6b4549 --- /dev/null +++ b/crates/server/src/error.rs @@ -0,0 +1,202 @@ +//! Error types of the server crate. + +use std::io; +use std::net::SocketAddr; + +use bgp_packet::errors::{BgpError, CeaseSubcode, DecodeError, FsmSubcode, OpenMsgSubcode}; +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", + }) + } +} + +/// Why a session ended abnormally. +#[derive(Debug, Error)] +pub enum SessionError { + #[error("connection failed: {0}")] + Io(#[from] io::Error), + #[error("received a message that could not be decoded: {0}")] + Decode(#[from] DecodeError), + #[error("could not encode a message: {0}")] + 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), + #[error("hold timer expired")] + HoldTimerExpired, + #[error("received {received} in the {state} state")] + UnexpectedMessage { + state: SessionState, + received: &'static str, + }, + #[error("refused the peer's OPEN: {0}")] + OpenRefused(BgpError), + #[error("peer announced AS{actual} but AS{expected} was configured")] + WrongPeerAs { expected: u32, actual: u32 }, + #[error("closed by connection collision resolution")] + Collision, +} + +impl SessionError { + /// The NOTIFICATION that tells the peer about this error (RFC 4271 §6), + /// 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::Decode(e) => { + return Some(NotificationMessage { + error: e.notification.clone(), + data: e.data.clone(), + }); + } + 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, + }), + Self::OpenRefused(error) => error.clone(), + Self::WrongPeerAs { .. } => BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs), + Self::Collision => BgpError::Cease(CeaseSubcode::ConnCollisionResolution), + }; + Some(NotificationMessage { + error, + data: vec![], + }) + } +} + +/// Why the server could not start. +#[derive(Debug, Error)] +pub enum ServerError { + #[error("failed to bind listener on {addr}: {source}")] + Bind { + addr: SocketAddr, + #[source] + source: io::Error, + }, + #[error("listener failed: {0}")] + Io(#[from] io::Error), + #[error("the local RIB stopped during startup")] + LocalRibStopped, +} + +/// Why a configuration was rejected. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum ConfigError { + #[error("invalid router_id {0:?}: must be a dotted-quad IPv4 address")] + RouterId(String), + #[error("local_asn must be non-zero")] + LocalAsn, + #[error("invalid IP address {0:?}")] + Address(String), + #[error("port {0} is out of range")] + Port(u32), + #[error("http_endpoint.port is required")] + HttpPort, + #[error("peer {peer:?}: {problem}")] + Peer { peer: String, problem: PeerProblem }, + #[error("originate {prefix:?}: {problem}")] + Originate { + prefix: String, + problem: OriginateProblem, + }, +} + +/// What is wrong with one `peers` entry. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum PeerProblem { + #[error("endpoint is missing")] + MissingEndpoint, + #[error("endpoint is invalid: {0}")] + Endpoint(Box), + #[error("remote_asn must be non-zero")] + RemoteAsn, + #[error("hold_time_seconds must be 0 or at least 3")] + HoldTime, + #[error("local_address {0:?} is not an IP address")] + LocalAddress(String), + #[error("local_address and endpoint are different address families")] + AddressFamilyMismatch, +} + +/// What is wrong with one `originate` entry. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum OriginateProblem { + #[error("{0}")] + Prefix(#[from] bgp_primitives::ip_prefix::PrefixError), + #[error("origin {0:?} must be IGP, EGP or INCOMPLETE")] + Origin(String), + #[error("as_path_prepend must be at most {max}")] + Prepend { max: u32 }, + #[error("{0}")] + Community(#[from] bgp_packet::community::CommunityError), + #[error("listed more than once")] + Duplicate, +} + +#[cfg(test)] +mod tests { + use bgp_packet::errors::MsgHeaderSubcode; + + use super::*; + + #[test] + fn each_session_error_names_the_notification_rfc_4271_assigns_it() { + let notification = |e: SessionError| e.notification().map(|n| (n.error, n.data)); + + assert_eq!( + notification(SessionError::Decode(DecodeError { + notification: BgpError::MsgHeader(MsgHeaderSubcode::BadMessageLength), + data: vec![0x13, 0x88], + detail: "too long".into(), + })), + Some(( + BgpError::MsgHeader(MsgHeaderSubcode::BadMessageLength), + vec![0x13, 0x88] + )) + ); + assert_eq!( + notification(SessionError::Timeout(SessionState::OpenSent)), + Some((BgpError::HoldTimer, vec![])) + ); + assert_eq!( + notification(SessionError::UnexpectedMessage { + state: SessionState::OpenConfirm, + received: "update" + }), + Some(( + BgpError::FsmError(FsmSubcode::UnexpectedOpenConfirm), + vec![] + )) + ); + assert_eq!( + notification(SessionError::WrongPeerAs { + expected: 1, + actual: 2 + }), + Some((BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs), vec![])) + ); + // A broken connection cannot carry a NOTIFICATION. + assert_eq!( + notification(SessionError::Io(io::ErrorKind::BrokenPipe.into())), + None + ); + } +} diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index 3887a48..6526f29 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -1,14 +1,13 @@ use std::net::{Ipv4Addr, SocketAddr}; use std::time::Duration; -use bgp_packet::errors::{BgpError, CeaseSubcode, FsmSubcode, OpenMsgSubcode}; -use bgp_packet::message::{BGP_HEADER_LEN, Message, NotificationMessage, PathAttribute}; +use bgp_packet::errors::{BgpError, DecodeError, MsgHeaderSubcode, OpenMsgSubcode}; +use bgp_packet::message::{BGP_HEADER_LEN, BGP_MAX_MSG_LEN, Message, PathAttribute}; 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 eyre::{Result, bail, eyre}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; @@ -19,6 +18,7 @@ use tracing::{info, warn}; use bgp_packet::message::{AS_TRANS, BGP4_VERSION}; use crate::collision::{CollisionTable, Direction}; +use crate::error::{SessionError, SessionState}; use crate::metrics::{ MESSAGES_RECEIVED, MESSAGES_SENT, SESSIONS_ESTABLISHED_TOTAL, message_type_name, }; @@ -54,7 +54,7 @@ pub async fn run_session( stream: TcpStream, peer_addr: SocketAddr, config: SessionConfig, -) -> Result<()> { +) -> Result<(), SessionError> { let local_addr = stream.local_addr()?; let (mut reader, mut writer) = stream.into_split(); @@ -65,38 +65,30 @@ pub async fn run_session( // 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, - &ParserContext::default(), + &pre_open, Duration::from_secs(240), + SessionState::OpenSent, ) .await; let peer_open = match peer_msg { Ok(Message::Open(o)) => o, Ok(other) => { - send_notification( - &mut writer, - BgpError::FsmError(FsmSubcode::UnexpectedOpenSent), - &ParserContext::default(), - ) - .await; - bail!("Expected OPEN in OpenSent state, got {:?}", other); - } - Err(e) => { - send_notification( - &mut writer, - BgpError::FsmError(FsmSubcode::UnexpectedOpenSent), - &ParserContext::default(), - ) - .await; - return Err(e); + 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), }; - if let Err(bgp_err) = validate_open(&peer_open, config.router_id) { - send_notification(&mut writer, bgp_err, &ParserContext::default()).await; - bail!("Peer {} sent invalid OPEN", peer_addr); + 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); } let negotiated_hold_time = config.hold_time_secs.min(peer_open.hold_time); @@ -127,13 +119,11 @@ pub async fn run_session( if let Some(expected) = config.expected_asn && peer_asn != expected { - send_notification( - &mut writer, - BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs), - &ctx, - ) - .await; - bail!("Peer {peer_addr} announced AS{peer_asn} but AS{expected} was configured"); + let error = SessionError::WrongPeerAs { + expected, + actual: peer_asn, + }; + return Err(abort(&mut writer, &ctx, error).await); } // Connection collision detection (RFC 4271 §6.8). From here on the @@ -168,7 +158,7 @@ pub async fn run_session( }); let confirm = tokio::select! { - msg = recv_with_timeout(&mut reader, &ctx, hold_duration) => msg, + 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; } @@ -177,18 +167,13 @@ pub async fn run_session( match confirm { Ok(Message::KeepAlive) => {} Ok(other) => { - send_notification( - &mut writer, - BgpError::FsmError(FsmSubcode::UnexpectedOpenConfirm), - &ctx, - ) - .await; - bail!("Expected KEEPALIVE in OpenConfirm state, got {:?}", other); - } - Err(e) => { - send_notification(&mut writer, BgpError::HoldTimer, &ctx).await; - return Err(e); + let error = SessionError::UnexpectedMessage { + state: SessionState::OpenConfirm, + 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"); @@ -254,7 +239,7 @@ async fn established_loop( peer_id: PeerId, adj_rib_in: &AdjRibIn, inputs: SessionInputs<'_>, -) -> Result<()> { +) -> Result<(), SessionError> { let SessionInputs { outbound, killed, @@ -282,7 +267,11 @@ async fn established_loop( loop { tokio::select! { result = read_message(reader, ctx) => { - match result? { + 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); } @@ -297,8 +286,11 @@ async fn established_loop( return Ok(()); } Message::Open(_) => { - send_notification(writer, BgpError::FsmError(FsmSubcode::UnexpectedEstablished), ctx).await; - bail!("Received OPEN in Established state from {}", peer_id); + 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); @@ -327,8 +319,7 @@ async fn established_loop( } _ = &mut hold_timer, if timers_enabled => { warn!(peer = %peer_id, "Hold timer expired"); - send_notification(writer, BgpError::HoldTimer, ctx).await; - bail!("Hold timer expired for peer {}", peer_id); + return Err(abort(writer, ctx, SessionError::HoldTimerExpired).await); } } } @@ -338,14 +329,27 @@ async fn established_loop( // Wire helpers // --------------------------------------------------------------------------- -async fn read_message(reader: &mut OwnedReadHalf, ctx: &ParserContext) -> Result { +async fn read_message( + reader: &mut OwnedReadHalf, + ctx: &ParserContext, +) -> Result { let mut header = [0u8; BGP_HEADER_LEN as usize]; reader.read_exact(&mut header).await?; - let total_len = u16::from_be_bytes([header[16], header[17]]) as usize; - if total_len < BGP_HEADER_LEN as usize { - bail!("BGP message length {total_len} below minimum"); + // 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()); } + let total_len = usize::from(length); let mut buf = vec![0u8; total_len]; buf[..BGP_HEADER_LEN as usize].copy_from_slice(&header); @@ -353,8 +357,7 @@ async fn read_message(reader: &mut OwnedReadHalf, ctx: &ParserContext) -> Result .read_exact(&mut buf[BGP_HEADER_LEN as usize..]) .await?; - let (_, msg) = - Message::from_wire(ctx, &buf).map_err(|e| eyre!("Failed to parse BGP message: {e:?}"))?; + let msg = Message::decode(ctx, &buf)?; MESSAGES_RECEIVED .with_label_values(&[message_type_name(&msg)]) .inc(); @@ -365,17 +368,18 @@ async fn recv_with_timeout( reader: &mut OwnedReadHalf, ctx: &ParserContext, timeout: Duration, -) -> Result { + state: SessionState, +) -> Result { tokio::time::timeout(timeout, read_message(reader, ctx)) .await - .map_err(|_| eyre!("Timed out waiting for BGP message"))? + .map_err(|_| SessionError::Timeout(state))? } async fn send_message( writer: &mut OwnedWriteHalf, msg: &Message, ctx: &ParserContext, -) -> Result<()> { +) -> Result<(), SessionError> { let mut buf = BytesMut::new(); msg.to_wire(ctx, &mut buf)?; writer.write_all(&buf).await?; @@ -385,16 +389,21 @@ async fn send_message( Ok(()) } -/// Best-effort NOTIFICATION send — errors are logged and swallowed since we -/// are already tearing down the session. -async fn send_notification(writer: &mut OwnedWriteHalf, error: BgpError, ctx: &ParserContext) { - let msg = Message::Notification(NotificationMessage { - error, - data: vec![], - }); - if let Err(e) = send_message(writer, &msg, ctx).await { - warn!("Failed to send NOTIFICATION: {e}"); +/// 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: @@ -403,15 +412,9 @@ async fn close_for_collision( writer: &mut OwnedWriteHalf, ctx: &ParserContext, peer_addr: SocketAddr, -) -> Result<()> { +) -> Result<(), SessionError> { info!(%peer_addr, "Closing connection: lost collision resolution"); - send_notification( - writer, - BgpError::Cease(CeaseSubcode::ConnCollisionResolution), - ctx, - ) - .await; - bail!("Connection to {peer_addr} closed by collision resolution"); + Err(abort(writer, ctx, SessionError::Collision).await) } // --------------------------------------------------------------------------- diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 3dd9ee0..ec48c69 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -3,6 +3,7 @@ pub mod collision; pub mod config; pub mod connection; pub mod dialer; +pub mod error; pub mod fsm; pub mod metrics; pub mod server; diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index a96082c..8f8fee1 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -2,7 +2,6 @@ use std::io; use std::net::SocketAddr; use std::sync::Arc; -use eyre::{Result, WrapErr}; use futures::stream::{FuturesUnordered, StreamExt}; use tokio::net::{TcpListener, TcpStream}; use tracing::{error, info}; @@ -14,6 +13,7 @@ use crate::collision::{CollisionTable, Direction}; use crate::config::Config; use crate::connection::handle_connection; use crate::dialer::spawn_dialers; +use crate::error::ServerError; use crate::fsm::SessionConfig; use crate::session::{ConfiguredPeer, SessionRegistry}; @@ -43,12 +43,12 @@ impl BgpServerBuilder { /// Binds all configured listen endpoints and returns a [`BgpServer`]. /// Separating bind from run lets callers (e.g. tests) inspect the actual /// bound addresses before the accept loop starts. - pub async fn bind(self) -> Result { + pub async fn bind(self) -> Result { let mut listeners = Vec::with_capacity(self.config.listen_endpoints.len()); for &addr in &self.config.listen_endpoints { let listener = TcpListener::bind(addr) .await - .wrap_err_with(|| format!("Failed to bind listener on {addr}"))?; + .map_err(|source| ServerError::Bind { addr, source })?; let bound_addr = listener.local_addr()?; info!(%bound_addr, "Listening for BGP connections"); listeners.push((bound_addr, listener)); @@ -60,7 +60,7 @@ impl BgpServerBuilder { local_rib .announce(route.clone()) .await - .wrap_err("Local RIB stopped during startup")?; + .map_err(|_| ServerError::LocalRibStopped)?; } Ok(BgpServer { listeners, @@ -82,7 +82,7 @@ impl BgpServerBuilder { } /// Convenience method: bind then run immediately. - pub async fn run(self) -> Result<()> { + pub async fn run(self) -> Result<(), ServerError> { self.bind().await?.run().await } } @@ -97,7 +97,7 @@ impl BgpServer { self.listeners.iter().map(|(addr, _)| *addr).collect() } - pub async fn run(self) -> Result<()> { + pub async fn run(self) -> Result<(), ServerError> { crate::metrics::init(); info!( router_id = %self.config.router_id, diff --git a/crates/server/src/tracing.rs b/crates/server/src/tracing.rs index 8d1ce92..5139809 100644 --- a/crates/server/src/tracing.rs +++ b/crates/server/src/tracing.rs @@ -1,7 +1,7 @@ use clap::Args; -use eyre::Result; use tracing::Level; use tracing_subscriber::EnvFilter; +use tracing_subscriber::filter::ParseError; /// Default targets that we enable logging for. const DEFAULT_TARGETS: [&str; 3] = ["bgp", "bgp_server", "bgp_web"]; @@ -17,10 +17,10 @@ pub struct TracingArgs { } impl TracingArgs { - pub fn install(&self) -> Result<()> { + pub fn install(&self) -> Result<(), ParseError> { let filter = DEFAULT_TARGETS.into_iter().try_fold( EnvFilter::builder().from_env_lossy(), - |filter, target| -> Result<_> { + |filter, target| -> Result<_, ParseError> { Ok(filter.add_directive(format!("{}={}", target, self.level).parse()?)) }, )?; diff --git a/crates/web/Cargo.toml b/crates/web/Cargo.toml index 34fa11b..42352ad 100644 --- a/crates/web/Cargo.toml +++ b/crates/web/Cargo.toml @@ -13,7 +13,6 @@ axum.workspace = true bgp-packet.workspace = true bgp-rib.workspace = true bgp-server.workspace = true -eyre.workspace = true futures.workspace = true prometheus.workspace = true serde.workspace = true diff --git a/crates/web/src/lib.rs b/crates/web/src/lib.rs index 017e276..46d09ff 100644 --- a/crates/web/src/lib.rs +++ b/crates/web/src/lib.rs @@ -19,7 +19,6 @@ use axum::response::{Html, IntoResponse, Redirect}; use axum::routing::get; use bgp_rib::{AdjRibIn, LocalRib}; use bgp_server::session::SessionRegistry; -use eyre::{Result, WrapErr}; use tokio::net::TcpListener; use tracing::info; @@ -108,15 +107,18 @@ pub fn router(state: AppState) -> Router { } /// Binds `listen` and serves the API until the task is dropped. -pub async fn serve(state: AppState, listen: SocketAddr) -> Result<()> { - let listener = TcpListener::bind(listen) - .await - .wrap_err_with(|| format!("Failed to bind HTTP listener on {listen}"))?; +pub async fn serve(state: AppState, listen: SocketAddr) -> std::io::Result<()> { + let listener = TcpListener::bind(listen).await.map_err(|e| { + std::io::Error::new( + e.kind(), + format!("failed to bind HTTP listener on {listen}: {e}"), + ) + })?; serve_on(state, listener).await } /// Serves the API on an already-bound listener (lets tests use port 0). -pub async fn serve_on(state: AppState, listener: TcpListener) -> Result<()> { +pub async fn serve_on(state: AppState, listener: TcpListener) -> std::io::Result<()> { info!(addr = %listener.local_addr()?, "HTTP API listening"); axum::serve(listener, router(state)).await?; Ok(())