Files
ferrix/crates/integration-tests/tests/basic_connection.rs
T
Rayhaan JaufeerallyandClaude Fable 5.1 16a4e3b640 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 <noreply@anthropic.com>
2026-10-01 21:18:35 +02:00

487 lines
16 KiB
Rust

use std::net::Ipv4Addr;
use std::time::Duration;
use bgp_integration_tests::test_peer::TestPeer;
use bgp_integration_tests::test_server::TestServer;
use bgp_packet::errors::{BgpError, CeaseSubcode, FsmSubcode, MsgHeaderSubcode, UpdateMsgSubcode};
use bgp_packet::message::{
Message, MpReachNlriPathAttribute, MpUnreachNlriPathAttribute, NlriNextHop,
NotificationMessage, PathAttribute, UpdateMessage,
};
use bgp_primitives::constants::{AddressFamilyId, SubsequentAfi};
use bgp_primitives::ip_prefix::IpPrefix;
use bgp_rib::RibEvent;
use eyre::Result;
const TIMEOUT: Duration = Duration::from_secs(5);
// ---------------------------------------------------------------------------
// Connectivity
// ---------------------------------------------------------------------------
#[tokio::test]
async fn test_server_starts() -> Result<()> {
let server = TestServer::start(65000).await?;
assert_ne!(server.addr.port(), 0);
Ok(())
}
#[tokio::test]
async fn test_server_accepts_tcp_connection() -> Result<()> {
let server = TestServer::start(65000).await?;
let _peer = TestPeer::connect(server.addr).await?;
Ok(())
}
#[tokio::test]
async fn test_server_accepts_multiple_connections() -> Result<()> {
let server = TestServer::start(65000).await?;
let mut peers = Vec::new();
for _ in 0..5 {
peers.push(TestPeer::connect(server.addr).await?);
}
assert_eq!(peers.len(), 5);
Ok(())
}
// ---------------------------------------------------------------------------
// OpenSent state
// ---------------------------------------------------------------------------
/// Server sends an OPEN immediately after TCP is established.
#[tokio::test]
async fn test_server_sends_open() -> Result<()> {
let server = TestServer::start(65000).await?;
let mut peer = TestPeer::connect(server.addr).await?;
let open = peer.recv_open().await?;
assert_eq!(open.version, 4);
assert_eq!(open.asn, 65000);
Ok(())
}
/// Server sends a NOTIFICATION and closes if we send garbage instead of OPEN.
#[tokio::test]
async fn test_bad_marker_does_not_crash_server() -> Result<()> {
let server = TestServer::start(65000).await?;
let mut bad = TestPeer::connect(server.addr).await?;
// Discard the server's OPEN, then send garbage.
let _ = bad.recv_open().await?;
bad.send_raw(&[0x00; 19]).await?;
drop(bad);
// New connections are still accepted.
let _good = TestPeer::connect(server.addr).await?;
Ok(())
}
/// Reads until a NOTIFICATION arrives, skipping KEEPALIVEs.
async fn expect_notification(peer: &mut TestPeer) -> Result<NotificationMessage> {
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<u8> {
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<u8>, BgpError, Vec<u8>)> = 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
// ---------------------------------------------------------------------------
/// Full OPEN/KEEPALIVE handshake reaches Established.
#[tokio::test]
async fn test_session_established() -> Result<()> {
let server = TestServer::start(65000).await?;
let mut peer = TestPeer::connect(server.addr).await?;
let server_open = peer
.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90)
.await?;
assert_eq!(server_open.version, 4);
Ok(())
}
/// After Established, the server sends periodic KEEPALIVEs.
#[tokio::test]
async fn test_server_sends_keepalives() -> Result<()> {
// Use a short hold time so the keepalive interval (hold/3 = 1 s) fires quickly.
let server = TestServer::start_with_hold_time(65000, 3).await?;
let mut peer = TestPeer::connect(server.addr).await?;
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 3).await?;
// We should receive a KEEPALIVE within ~1 second (hold/3).
let msg = tokio::time::timeout(Duration::from_secs(3), peer.recv()).await??;
assert!(
matches!(msg, Message::KeepAlive),
"Expected KEEPALIVE, got {:?}",
msg
);
Ok(())
}
/// Server closes with a NOTIFICATION when the hold timer expires.
#[tokio::test]
async fn test_hold_timer_expires() -> Result<()> {
// hold_time = 3 s → server expects a KEEPALIVE within 3 s.
let server = TestServer::start_with_hold_time(65000, 3).await?;
let mut peer = TestPeer::connect(server.addr).await?;
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 3).await?;
// Don't send any keepalives. The server will send keepalives until its own
// hold timer fires and it sends a NOTIFICATION. Drain messages until we
// see it (or the outer timeout fires).
let notif = tokio::time::timeout(Duration::from_secs(6), async {
loop {
match peer.recv().await? {
Message::Notification(n) => return Ok::<_, eyre::Error>(n),
Message::KeepAlive => {} // server keepalive — keep waiting
other => eyre::bail!("Unexpected message: {:?}", other),
}
}
})
.await??;
assert!(
matches!(notif.error, BgpError::HoldTimer),
"Expected HoldTimer NOTIFICATION, got {:?}",
notif
);
Ok(())
}
/// A peer-sent NOTIFICATION is handled gracefully (session closes cleanly).
#[tokio::test]
async fn test_peer_sends_notification() -> Result<()> {
let server = TestServer::start(65000).await?;
let mut peer = TestPeer::connect(server.addr).await?;
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90)
.await?;
peer.send(&Message::Notification(NotificationMessage {
error: BgpError::Cease(CeaseSubcode::AdminShutdown),
data: vec![],
}))
.await?;
// Server should close the connection cleanly.
tokio::time::sleep(Duration::from_millis(100)).await;
assert!(peer.is_closed().await);
Ok(())
}
// ---------------------------------------------------------------------------
// UPDATE
// ---------------------------------------------------------------------------
/// An UPDATE message is accepted without crashing the server.
#[tokio::test]
async fn test_server_accepts_update() -> Result<()> {
let server = TestServer::start(65000).await?;
let mut peer = TestPeer::connect(server.addr).await?;
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90)
.await?;
let update = Message::Update(UpdateMessage {
withdrawn_routes: vec![],
path_attributes: vec![],
// 10.0.0.0/8 — one byte of prefix for a /8.
announced_routes: vec![IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?],
});
peer.send(&update).await?;
// Server should stay up — we can still exchange a keepalive.
peer.send(&Message::KeepAlive).await?;
Ok(())
}
/// An IPv6 UPDATE via MP_REACH_NLRI is accepted without crashing the server.
#[tokio::test]
async fn test_server_accepts_ipv6_update() -> Result<()> {
let server = TestServer::start(65000).await?;
let mut peer = TestPeer::connect(server.addr).await?;
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90)
.await?;
// 2001:db8::/32
let prefix = IpPrefix::new(AddressFamilyId::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?;
let update = Message::Update(UpdateMessage {
withdrawn_routes: vec![],
path_attributes: vec![PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
afi: AddressFamilyId::Ipv6,
safi: SubsequentAfi::Unicast,
next_hop: NlriNextHop::Ipv6("2001:db8::1".parse().unwrap()),
prefixes: vec![prefix],
})],
announced_routes: vec![],
});
peer.send(&update).await?;
// Server should stay up — we can still exchange a keepalive.
peer.send(&Message::KeepAlive).await?;
Ok(())
}
// ---------------------------------------------------------------------------
// Adj-RIB-In introspection
// ---------------------------------------------------------------------------
/// IPv4 prefix announced via the legacy NLRI field appears in the RIB.
#[tokio::test]
async fn test_rib_records_ipv4_announce() -> Result<()> {
let server = TestServer::start(65000).await?;
let mut events = server.adj_rib_in.subscribe();
let mut peer = TestPeer::connect(server.addr).await?;
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90)
.await?;
let prefix = IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?;
peer.send(&Message::Update(UpdateMessage {
withdrawn_routes: vec![],
path_attributes: vec![],
announced_routes: vec![prefix.clone()],
}))
.await?;
let event = tokio::time::timeout(Duration::from_secs(1), events.recv()).await??;
assert!(
matches!(&event, RibEvent::Announced { prefix: p, .. } if p == &prefix),
"Expected Announced for 10.0.0.0/8, got {event:?}",
);
let snapshot = server.adj_rib_in.snapshot().await;
assert!(
snapshot.iter().any(|e| e.prefix == prefix),
"10.0.0.0/8 missing from RIB snapshot",
);
Ok(())
}
/// Withdrawn IPv4 prefix is removed from the RIB.
#[tokio::test]
async fn test_rib_records_ipv4_withdraw() -> Result<()> {
let server = TestServer::start(65000).await?;
let mut events = server.adj_rib_in.subscribe();
let mut peer = TestPeer::connect(server.addr).await?;
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90)
.await?;
let prefix = IpPrefix::new(AddressFamilyId::Ipv4, vec![10], 8)?;
// Announce first.
peer.send(&Message::Update(UpdateMessage {
withdrawn_routes: vec![],
path_attributes: vec![],
announced_routes: vec![prefix.clone()],
}))
.await?;
tokio::time::timeout(Duration::from_secs(1), events.recv()).await??;
// Now withdraw.
peer.send(&Message::Update(UpdateMessage {
withdrawn_routes: vec![prefix.clone()],
path_attributes: vec![],
announced_routes: vec![],
}))
.await?;
let event = tokio::time::timeout(Duration::from_secs(1), events.recv()).await??;
assert!(
matches!(&event, RibEvent::Withdrawn { prefix: p, .. } if p == &prefix),
"Expected Withdrawn for 10.0.0.0/8, got {event:?}",
);
let snapshot = server.adj_rib_in.snapshot().await;
assert!(
!snapshot.iter().any(|e| e.prefix == prefix),
"10.0.0.0/8 should have been removed from RIB snapshot",
);
Ok(())
}
/// IPv6 prefix announced via MP_REACH_NLRI appears in the RIB.
#[tokio::test]
async fn test_rib_records_ipv6_announce() -> Result<()> {
let server = TestServer::start(65000).await?;
let mut events = server.adj_rib_in.subscribe();
let mut peer = TestPeer::connect(server.addr).await?;
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90)
.await?;
let prefix = IpPrefix::new(AddressFamilyId::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?;
peer.send(&Message::Update(UpdateMessage {
withdrawn_routes: vec![],
path_attributes: vec![PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
afi: AddressFamilyId::Ipv6,
safi: SubsequentAfi::Unicast,
next_hop: NlriNextHop::Ipv6("2001:db8::1".parse().unwrap()),
prefixes: vec![prefix.clone()],
})],
announced_routes: vec![],
}))
.await?;
let event = tokio::time::timeout(Duration::from_secs(1), events.recv()).await??;
assert!(
matches!(&event, RibEvent::Announced { prefix: p, .. } if p == &prefix),
"Expected Announced for 2001:db8::/32, got {event:?}",
);
let snapshot = server.adj_rib_in.snapshot().await;
assert!(
snapshot.iter().any(|e| e.prefix == prefix),
"2001:db8::/32 missing from RIB snapshot",
);
Ok(())
}
/// IPv6 prefix withdrawn via MP_UNREACH_NLRI is removed from the RIB.
#[tokio::test]
async fn test_rib_records_ipv6_withdraw() -> Result<()> {
let server = TestServer::start(65000).await?;
let mut events = server.adj_rib_in.subscribe();
let mut peer = TestPeer::connect(server.addr).await?;
peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90)
.await?;
let prefix = IpPrefix::new(AddressFamilyId::Ipv6, vec![0x20, 0x01, 0x0d, 0xb8], 32)?;
// Announce first.
peer.send(&Message::Update(UpdateMessage {
withdrawn_routes: vec![],
path_attributes: vec![PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
afi: AddressFamilyId::Ipv6,
safi: SubsequentAfi::Unicast,
next_hop: NlriNextHop::Ipv6("2001:db8::1".parse().unwrap()),
prefixes: vec![prefix.clone()],
})],
announced_routes: vec![],
}))
.await?;
tokio::time::timeout(Duration::from_secs(1), events.recv()).await??;
// Now withdraw via MP_UNREACH_NLRI.
peer.send(&Message::Update(UpdateMessage {
withdrawn_routes: vec![],
path_attributes: vec![PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute {
afi: AddressFamilyId::Ipv6,
safi: SubsequentAfi::Unicast,
prefixes: vec![prefix.clone()],
})],
announced_routes: vec![],
}))
.await?;
let event = tokio::time::timeout(Duration::from_secs(1), events.recv()).await??;
assert!(
matches!(&event, RibEvent::Withdrawn { prefix: p, .. } if p == &prefix),
"Expected Withdrawn for 2001:db8::/32, got {event:?}",
);
let snapshot = server.adj_rib_in.snapshot().await;
assert!(
!snapshot.iter().any(|e| e.prefix == prefix),
"2001:db8::/32 should have been removed from RIB snapshot",
);
Ok(())
}