diff --git a/src/bin/rust-bfdd.rs b/src/bin/rust-bfdd.rs index f1b92ab..843160b 100644 --- a/src/bin/rust-bfdd.rs +++ b/src/bin/rust-bfdd.rs @@ -47,6 +47,31 @@ struct Args { peers: Vec, } +// RFC 5881 section 4: the source port MUST be in the range 49152 to 65535. +const SOURCE_PORT_RANGE: std::ops::RangeInclusive = 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 { + 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; #[derive(Debug)] @@ -70,6 +95,7 @@ impl Error for BfdError {} #[derive(Debug)] struct BfdSessionState { control_sock: Arc, // immutable + local_ip: IpAddr, // immutable and Copy peer_addr: IpAddr, // immutable and Copy session_state: Atomic, // mutable by any thread remote_session_state: Atomic, // mutable by rx thread @@ -101,6 +127,9 @@ impl BfdSessionState { fn control_sock(&self) -> Arc { self.control_sock.clone() } + fn local_ip(&self) -> IpAddr { + self.local_ip + } fn peer_addr(&self) -> IpAddr { self.peer_addr } @@ -219,7 +248,7 @@ struct BfdSession { impl From<&BfdSessionState> for BfdSessionStats { fn from(state: &BfdSessionState) -> Self { Self { - local_ip: state.control_sock().local_addr().unwrap().ip(), + local_ip: state.local_ip(), remote_ip: state.peer_addr(), local_discr: state.local_discr(), remote_discr: state.remote_discr(), @@ -264,7 +293,9 @@ impl BfdSessionHandle { remote_addr: IpAddr, local_discr: BfdDiscriminator, event_dispatcher: EventDispatcherHandle, - ) -> Self { + ) -> io::Result { + let control_sock = bind_control_socket(local_addr).await?; + let (tx, rx) = mpsc::channel(32); let new_self = Self { tx }; let periodic = @@ -274,15 +305,15 @@ impl BfdSessionHandle { rx, new_self.tx.clone(), periodic, + control_sock, local_addr, remote_addr, local_discr, event_dispatcher, - ) - .await; + ); tokio::spawn(async move { session.run().await }); - new_self + Ok(new_self) } async fn transmit_control_packet( &self, @@ -382,10 +413,11 @@ impl BfdPeriodicSender { } impl BfdSession { - async fn new( + fn new( rx: mpsc::Receiver, tx: mpsc::Sender, periodic: BfdPeriodicSenderHandle, + control_sock: UdpSocket, local_addr: IpAddr, remote_addr: IpAddr, local_discr: BfdDiscriminator, @@ -393,12 +425,6 @@ impl BfdSession { ) -> Self { 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 // .connect(SocketAddr::new(remote_addr, CONTROL_PORT)) // .await?; @@ -413,6 +439,7 @@ impl BfdSession { event_dispatcher, state: BfdSessionState { control_sock: Arc::new(control_sock), + local_ip: local_addr, peer_addr: remote_addr, session_state: Atomic::new(BfdState::default()), remote_session_state: Atomic::new(BfdState::default()), @@ -448,7 +475,7 @@ impl BfdSession { .dispatch(EventMessage::StateChange(StateChangeEvent { local_discr: self.state.local_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(), from_state: self.state.session_state(), to_state: new_state, @@ -977,7 +1004,7 @@ impl Bfdd { event_dispatcher, }) } - async fn start_peers(&mut self) { + async fn start_peers(&mut self) -> io::Result<()> { let mut rng = rand::thread_rng(); for peer in &self.peers { let mut local_discr = BfdDiscriminator(0); @@ -990,12 +1017,13 @@ impl Bfdd { local_discr, self.event_dispatcher.clone(), ) - .await; + .await?; self.sessions_by_ip .insert((self.local.ip(), peer.ip()), local_discr); self.sessions_by_discr.insert(local_discr, session); } + Ok(()) } fn session_for_packet( &self, @@ -1012,11 +1040,10 @@ impl Bfdd { // 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). let our_disc = if packet.your_disc == BfdDiscriminator(0) { - if let Some(discr) = self - .sessions_by_ip - .get(&(self.control_socket.local_addr().unwrap().ip(), addr.ip())) - { - // TODO: Probably we should destroy the old session here? + if let Some(discr) = self.sessions_by_ip.get(&(self.local.ip(), addr.ip())) { + // A restarted peer starts over with Your Discriminator of zero. We deliberately keep the existing + // 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. debug!( "Found session for unknown discriminator from {}: {}", addr.ip(), @@ -1046,7 +1073,7 @@ impl Bfdd { } async fn run(mut self) -> io::Result<()> { info!("rust-bfd starting up"); - self.start_peers().await; + self.start_peers().await?; info!( "configured sessions: {}", future::join_all( @@ -1058,33 +1085,39 @@ impl Bfdd { .join("\n\t") ); - let rx_task = - task::spawn(async move { - let mut buf = [0; 1024]; - loop { - // 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. - // - // 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. - let (len, addr) = self.control_socket.recv_from(&mut buf).await.unwrap(); // TODO: fallibility? - if let Ok((_leftover, packet)) = BfdPacket::parse(&buf[..len]) { - debug!("rx packet ({:?}): {:?}", addr.ip(), packet); - self.rx_packet(addr, packet).await - } else { - warn!("Failed to parse packet"); - } + let rx_task = task::spawn(async move { + let mut buf = [0; 1024]; + loop { + // 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. + // + // 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. + match self.control_socket.recv_from(&mut buf).await { + Ok((len, addr)) => { + if let Ok((_leftover, packet)) = BfdPacket::parse(&buf[..len]) { + debug!("rx packet ({:?}): {:?}", addr.ip(), packet); + self.rx_packet(addr, packet).await + } else { + warn!("Failed to parse packet"); + } - info!( - "configured sessions: {}", - future::join_all(self.sessions_by_discr.values().map(|session| async { - session.get_stats().await.unwrap().to_string() - })) - .await - .join("\n\t") - ); + info!( + "configured sessions: {}", + future::join_all(self.sessions_by_discr.values().map( + |session| async { session.get_stats().await.unwrap().to_string() } + )) + .await + .join("\n\t") + ); + } + Err(e) => { + warn!("Error receiving from socket {e:?}"); + continue; + } } - }); + } + }); tokio::join!(rx_task) .0