ktims/rust-bfd
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Bind session sockets without panicking

Browse Source 
Select a random unused source port in the RFC 5881 range (retrying on EADDRINUSE) instead of unwrapping a pure-random bind. Propagate bind errors out of session start-up, store the local IP on the session instead of unwrapping local_addr(), and log (rather than panic on) receive errors in the rx loop.
This commit is contained in:
ktims committed 2026-10-22 15:41:08 -07:00
1 parent c7a6ed16e4
commit 0bd4f8766a
1 file changed
+79 -46
Download Patch File Download Diff File Expand all files Collapse all files
src/bin/rust-bfdd.rs
+79 -46
View File
@@ -47,6 +47,31 @@ struct Args {
peers: Vec<IpAddr>, peers: Vec<IpAddr>,
} }
// RFC 5881 section 4: the source port MUST be in the range 49152 to 65535.
const SOURCE_PORT_RANGE: std::ops::RangeInclusive<u16> = 49152..=65535;
const SOURCE_PORT_BIND_ATTEMPTS: usize = 32;
/// Bind a UDP socket on a random unused port in the RFC 5881 source port range, with TTL/hop limit 255.
async fn bind_control_socket(local_addr: IpAddr) -> io::Result<UdpSocket> {
for _ in 0..SOURCE_PORT_BIND_ATTEMPTS {
let port = rand::thread_rng().gen_range(SOURCE_PORT_RANGE);
match UdpSocket::bind(SocketAddr::new(local_addr, port)).await {
Ok(sock) => {
set_ttl_or_hop_limit(&sock, 255)?;
return Ok(sock);
}
Err(e) if e.kind() == ErrorKind::AddrInUse => continue,
Err(e) => return Err(e),
}
}
Err(io::Error::new(
ErrorKind::AddrInUse,
format!(
"no free source port found for {local_addr} after {SOURCE_PORT_BIND_ATTEMPTS} attempts"
),
))
}
const ORDERING: Ordering = Ordering::Relaxed; const ORDERING: Ordering = Ordering::Relaxed;
#[derive(Debug)] #[derive(Debug)]
@@ -70,6 +95,7 @@ impl Error for BfdError {}
#[derive(Debug)] #[derive(Debug)]
struct BfdSessionState { struct BfdSessionState {
control_sock: Arc<UdpSocket>, // immutable control_sock: Arc<UdpSocket>, // immutable
local_ip: IpAddr, // immutable and Copy
peer_addr: IpAddr, // immutable and Copy peer_addr: IpAddr, // immutable and Copy
session_state: Atomic<BfdState>, // mutable by any thread session_state: Atomic<BfdState>, // mutable by any thread
remote_session_state: Atomic<BfdState>, // mutable by rx thread remote_session_state: Atomic<BfdState>, // mutable by rx thread
@@ -101,6 +127,9 @@ impl BfdSessionState {
fn control_sock(&self) -> Arc<UdpSocket> { fn control_sock(&self) -> Arc<UdpSocket> {
self.control_sock.clone() self.control_sock.clone()
} }
fn local_ip(&self) -> IpAddr {
self.local_ip
}
fn peer_addr(&self) -> IpAddr { fn peer_addr(&self) -> IpAddr {
self.peer_addr self.peer_addr
} }
@@ -219,7 +248,7 @@ struct BfdSession {
impl From<&BfdSessionState> for BfdSessionStats { impl From<&BfdSessionState> for BfdSessionStats {
fn from(state: &BfdSessionState) -> Self { fn from(state: &BfdSessionState) -> Self {
Self { Self {
local_ip: state.control_sock().local_addr().unwrap().ip(), local_ip: state.local_ip(),
remote_ip: state.peer_addr(), remote_ip: state.peer_addr(),
local_discr: state.local_discr(), local_discr: state.local_discr(),
remote_discr: state.remote_discr(), remote_discr: state.remote_discr(),
@@ -264,7 +293,9 @@ impl BfdSessionHandle {
remote_addr: IpAddr, remote_addr: IpAddr,
local_discr: BfdDiscriminator, local_discr: BfdDiscriminator,
event_dispatcher: EventDispatcherHandle, event_dispatcher: EventDispatcherHandle,
) -> Self { ) -> io::Result<Self> {
let control_sock = bind_control_socket(local_addr).await?;
let (tx, rx) = mpsc::channel(32); let (tx, rx) = mpsc::channel(32);
let new_self = Self { tx }; let new_self = Self { tx };
let periodic = let periodic =
@@ -274,15 +305,15 @@ impl BfdSessionHandle {
rx, rx,
new_self.tx.clone(), new_self.tx.clone(),
periodic, periodic,
control_sock,
local_addr, local_addr,
remote_addr, remote_addr,
local_discr, local_discr,
event_dispatcher, event_dispatcher,
) );
.await;
tokio::spawn(async move { session.run().await }); tokio::spawn(async move { session.run().await });
new_self Ok(new_self)
} }
async fn transmit_control_packet( async fn transmit_control_packet(
&self, &self,
@@ -382,10 +413,11 @@ impl BfdPeriodicSender {
} }
impl BfdSession { impl BfdSession {
async fn new( fn new(
rx: mpsc::Receiver<SessionControlCommand>, rx: mpsc::Receiver<SessionControlCommand>,
tx: mpsc::Sender<SessionControlCommand>, tx: mpsc::Sender<SessionControlCommand>,
periodic: BfdPeriodicSenderHandle, periodic: BfdPeriodicSenderHandle,
control_sock: UdpSocket,
local_addr: IpAddr, local_addr: IpAddr,
remote_addr: IpAddr, remote_addr: IpAddr,
local_discr: BfdDiscriminator, local_discr: BfdDiscriminator,
@@ -393,12 +425,6 @@ impl BfdSession {
) -> Self { ) -> Self {
let mut rng = rand::thread_rng(); let mut rng = rand::thread_rng();
//TODO: select a random unused port instead of pure random
let source_port: u16 = rng.gen_range(49152..=65535);
let control_sock = UdpSocket::bind(SocketAddr::new(local_addr, source_port))
.await
.unwrap();
set_ttl_or_hop_limit(&control_sock, 255).unwrap();
// control_sock // control_sock
// .connect(SocketAddr::new(remote_addr, CONTROL_PORT)) // .connect(SocketAddr::new(remote_addr, CONTROL_PORT))
// .await?; // .await?;
@@ -413,6 +439,7 @@ impl BfdSession {
event_dispatcher, event_dispatcher,
state: BfdSessionState { state: BfdSessionState {
control_sock: Arc::new(control_sock), control_sock: Arc::new(control_sock),
local_ip: local_addr,
peer_addr: remote_addr, peer_addr: remote_addr,
session_state: Atomic::new(BfdState::default()), session_state: Atomic::new(BfdState::default()),
remote_session_state: Atomic::new(BfdState::default()), remote_session_state: Atomic::new(BfdState::default()),
@@ -448,7 +475,7 @@ impl BfdSession {
.dispatch(EventMessage::StateChange(StateChangeEvent { .dispatch(EventMessage::StateChange(StateChangeEvent {
local_discr: self.state.local_discr(), local_discr: self.state.local_discr(),
remote_discr: self.state.remote_discr(), remote_discr: self.state.remote_discr(),
local_ip: self.state.control_sock().local_addr().unwrap().ip(), //TODO: shouldn't unwrap here, it will crash the program local_ip: self.state.local_ip(),
remote_ip: self.state.peer_addr(), remote_ip: self.state.peer_addr(),
from_state: self.state.session_state(), from_state: self.state.session_state(),
to_state: new_state, to_state: new_state,
@@ -977,7 +1004,7 @@ impl Bfdd {
event_dispatcher, event_dispatcher,
}) })
} }
async fn start_peers(&mut self) { async fn start_peers(&mut self) -> io::Result<()> {
let mut rng = rand::thread_rng(); let mut rng = rand::thread_rng();
for peer in &self.peers { for peer in &self.peers {
let mut local_discr = BfdDiscriminator(0); let mut local_discr = BfdDiscriminator(0);
@@ -990,12 +1017,13 @@ impl Bfdd {
local_discr, local_discr,
self.event_dispatcher.clone(), self.event_dispatcher.clone(),
) )
.await; .await?;
self.sessions_by_ip self.sessions_by_ip
.insert((self.local.ip(), peer.ip()), local_discr); .insert((self.local.ip(), peer.ip()), local_discr);
self.sessions_by_discr.insert(local_discr, session); self.sessions_by_discr.insert(local_discr, session);
} }
Ok(())
} }
fn session_for_packet( fn session_for_packet(
&self, &self,
@@ -1012,11 +1040,10 @@ impl Bfdd {
// source address field can change or the interface over which the packets are received can change, but // source address field can change or the interface over which the packets are received can change, but
// the packets will still be associated with the proper session). // the packets will still be associated with the proper session).
let our_disc = if packet.your_disc == BfdDiscriminator(0) { let our_disc = if packet.your_disc == BfdDiscriminator(0) {
if let Some(discr) = self if let Some(discr) = self.sessions_by_ip.get(&(self.local.ip(), addr.ip())) {
.sessions_by_ip // A restarted peer starts over with Your Discriminator of zero. We deliberately keep the existing
.get(&(self.control_socket.local_addr().unwrap().ip(), addr.ip())) // session: the RFC 5880 section 6.8.6 state machine takes it Down (diag: Neighbor signaled session
{ // down) and bfd.RemoteDiscr is relearned from the new My Discriminator.
// TODO: Probably we should destroy the old session here?
debug!( debug!(
"Found session for unknown discriminator from {}: {}", "Found session for unknown discriminator from {}: {}",
addr.ip(), addr.ip(),
@@ -1046,7 +1073,7 @@ impl Bfdd {
} }
async fn run(mut self) -> io::Result<()> { async fn run(mut self) -> io::Result<()> {
info!("rust-bfd starting up"); info!("rust-bfd starting up");
self.start_peers().await; self.start_peers().await?;
info!( info!(
"configured sessions: {}", "configured sessions: {}",
future::join_all( future::join_all(
@@ -1058,33 +1085,39 @@ impl Bfdd {
.join("\n\t") .join("\n\t")
); );
let rx_task = let rx_task = task::spawn(async move {
task::spawn(async move { let mut buf = [0; 1024];
let mut buf = [0; 1024]; loop {
loop { // TODO: All received BFD Control packets that are demultiplexed to the session MUST be discarded if
// TODO: All received BFD Control packets that are demultiplexed to the session MUST be discarded if // the received TTL or Hop Limit is not equal to 255. Need raw sockets for this.
// the received TTL or Hop Limit is not equal to 255. Need raw sockets for this. //
// // TODO: Maybe it is okay to receive in the session-specific socket instead of this global one? But
// TODO: Maybe it is okay to receive in the session-specific socket instead of this global one? But // the RFC mentions that we must demux based on discriminator only.
// the RFC mentions that we must demux based on discriminator only. match self.control_socket.recv_from(&mut buf).await {
let (len, addr) = self.control_socket.recv_from(&mut buf).await.unwrap(); // TODO: fallibility? Ok((len, addr)) => {
if let Ok((_leftover, packet)) = BfdPacket::parse(&buf[..len]) { if let Ok((_leftover, packet)) = BfdPacket::parse(&buf[..len]) {
debug!("rx packet ({:?}): {:?}", addr.ip(), packet); debug!("rx packet ({:?}): {:?}", addr.ip(), packet);
self.rx_packet(addr, packet).await self.rx_packet(addr, packet).await
} else { } else {
warn!("Failed to parse packet"); warn!("Failed to parse packet");
} }
info!( info!(
"configured sessions: {}", "configured sessions: {}",
future::join_all(self.sessions_by_discr.values().map(|session| async { future::join_all(self.sessions_by_discr.values().map(
session.get_stats().await.unwrap().to_string() |session| async { session.get_stats().await.unwrap().to_string() }
})) ))
.await .await
.join("\n\t") .join("\n\t")
); );
}
Err(e) => {
warn!("Error receiving from socket {e:?}");
continue;
}
} }
}); }
});
tokio::join!(rx_task) tokio::join!(rx_task)
.0 .0