This commit is contained in:
ktims committed 2026-10-22 15:47:16 -07:00
1 parent dcbf25ebe0
commit 0284ee17b1
3 files changed
+53 -63

No files matched your search

+14 -8
View File
@@ -1,4 +1,10 @@
use std::{ffi::{OsStr, OsString}, fmt::{self, Display}, io::Cursor, ops::{Div, Mul}, time}; use std::{
ffi::OsStr,
fmt::{self, Display},
io::Cursor,
ops::{Div, Mul},
time,
};
use bytemuck::NoUninit; use bytemuck::NoUninit;
use byteorder::{BigEndian, WriteBytesExt}; use byteorder::{BigEndian, WriteBytesExt};
@@ -56,13 +62,13 @@ impl Display for BfdState {
} }
} }
impl Into<&OsStr> for BfdState { impl From<BfdState> for &OsStr {
fn into(self) -> &'static OsStr { fn from(val: BfdState) -> Self {
match self { match val {
Self::AdminDown => OsStr::new("AdminDown"), BfdState::AdminDown => OsStr::new("AdminDown"),
Self::Init => OsStr::new("Init"), BfdState::Init => OsStr::new("Init"),
Self::Down => OsStr::new("Down"), BfdState::Down => OsStr::new("Down"),
Self::Up => OsStr::new("Up"), BfdState::Up => OsStr::new("Up"),
} }
} }
} }
+29 -50
View File
@@ -15,12 +15,11 @@ use std::{
}; };
// Necessary until a hop_limit method is added to UdpSocket // Necessary until a hop_limit method is added to UdpSocket
use nix::sys::socket::{setsockopt, sockopt};
use tokio::sync::{oneshot, Mutex}; use tokio::sync::{oneshot, Mutex};
use env_logger::Env; use env_logger::Env;
use log::{debug, error, info, warn}; use log::{debug, info, warn};
use nom_derive::Parse; use nom_derive::Parse;
use rand::prelude::*; use rand::prelude::*;
@@ -302,13 +301,10 @@ impl BfdSessionHandle {
let (tx, rx) = mpsc::channel(32); let (tx, rx) = mpsc::channel(32);
let new_self = Self { tx }; let new_self = Self { tx };
let periodic =
BfdPeriodicSenderHandle::new(new_self.clone(), time::Duration::from_secs(1).into());
let mut session = BfdSession::new( let mut session = BfdSession::new(
rx, rx,
new_self.tx.clone(), new_self.clone(),
periodic,
control_sock, control_sock,
local_addr, local_addr,
remote_addr, remote_addr,
@@ -378,7 +374,7 @@ impl BfdPeriodicSender {
loop { loop {
tokio::select! { tokio::select! {
_ = clock.tick(), if running == true => { _ = clock.tick(), if running => {
let jitter = time::Duration::from_micros( let jitter = time::Duration::from_micros(
rand::thread_rng() rand::thread_rng()
.gen_range(0..clock.period().as_micros() / 3) .gen_range(0..clock.period().as_micros() / 3)
@@ -401,8 +397,8 @@ impl BfdPeriodicSender {
let base_interval: Duration = (i * 3 / 4).into(); let base_interval: Duration = (i * 3 / 4).into();
debug!("Updating base interval to {}ms (jittering {} to {}ms)", base_interval.as_millis(), base_interval.as_millis(), base_interval.as_millis() + base_interval.as_millis() / 3); debug!("Updating base interval to {}ms (jittering {} to {}ms)", base_interval.as_millis(), base_interval.as_millis(), base_interval.as_millis() + base_interval.as_millis() / 3);
clock = time::interval_at( clock = time::interval_at(
time::Instant::now() + base_interval.into(), time::Instant::now() + base_interval,
base_interval.into(), base_interval,
); );
}, },
PeriodicControlCommand::SendPollResponse => { PeriodicControlCommand::SendPollResponse => {
@@ -419,8 +415,7 @@ impl BfdPeriodicSender {
impl BfdSession { impl BfdSession {
fn new( fn new(
rx: mpsc::Receiver<SessionControlCommand>, rx: mpsc::Receiver<SessionControlCommand>,
tx: mpsc::Sender<SessionControlCommand>, handle: BfdSessionHandle,
periodic: BfdPeriodicSenderHandle,
control_sock: UdpSocket, control_sock: UdpSocket,
local_addr: IpAddr, local_addr: IpAddr,
remote_addr: IpAddr, remote_addr: IpAddr,
@@ -428,6 +423,8 @@ impl BfdSession {
event_dispatcher: EventDispatcherHandle, event_dispatcher: EventDispatcherHandle,
) -> Self { ) -> Self {
let mut rng = rand::thread_rng(); let mut rng = rand::thread_rng();
let periodic =
BfdPeriodicSenderHandle::new(handle.clone(), time::Duration::from_secs(1).into());
let configured_desired_min_tx = BfdInterval::from_millis(300); let configured_desired_min_tx = BfdInterval::from_millis(300);
let slow_desired_min_tx = let slow_desired_min_tx =
std::cmp::max(configured_desired_min_tx, BfdInterval::from_secs(1)); std::cmp::max(configured_desired_min_tx, BfdInterval::from_secs(1));
@@ -439,7 +436,7 @@ impl BfdSession {
// Incoming packets will come over the channel from the mux, since they don't send to the reciprocal port // Incoming packets will come over the channel from the mux, since they don't send to the reciprocal port
Self { Self {
rx, rx,
tx, tx: handle.tx,
periodic, periodic,
rx_watchdog: None, rx_watchdog: None,
cur_interval: BfdInterval::from_secs(1), cur_interval: BfdInterval::from_secs(1),
@@ -499,13 +496,8 @@ impl BfdSession {
*self.state.last_state_change.lock().await = Instant::now(); *self.state.last_state_change.lock().await = Instant::now();
match (self.state.session_state(), new_state) { match (self.state.session_state(), new_state) {
(BfdState::Up, BfdState::AdminDown | BfdState::Down | BfdState::Init) => { (BfdState::Up, BfdState::AdminDown | BfdState::Down | BfdState::Init) => {
if self.rx_watchdog.is_some() { if let Some(watchdog) = self.rx_watchdog.as_mut() {
self.rx_watchdog watchdog.send(WatchdogReset::Terminate).await.unwrap();
.as_mut()
.unwrap()
.send(WatchdogReset::Terminate)
.await
.unwrap();
}; };
// When bfd.SessionState is not Up, the system MUST set bfd.DesiredMinTxInterval to a value of not // When bfd.SessionState is not Up, the system MUST set bfd.DesiredMinTxInterval to a value of not
// less than one second (1,000,000 microseconds). This is intended to ensure that the bandwidth // less than one second (1,000,000 microseconds). This is intended to ensure that the bandwidth
@@ -546,7 +538,7 @@ impl BfdSession {
self.receive_control_packet(&packet).await; self.receive_control_packet(&packet).await;
} }
SessionControlCommand::GetSessionStats { respond_to } => { SessionControlCommand::GetSessionStats { respond_to } => {
let mut stats = let stats =
BfdSessionStats::from(&self.state).with_base_interval(self.cur_interval); BfdSessionStats::from(&self.state).with_base_interval(self.cur_interval);
respond_to.send(stats).unwrap() respond_to.send(stats).unwrap()
} }
@@ -589,20 +581,17 @@ impl BfdSession {
let dest = self.state.peer_addr; let dest = self.state.peer_addr;
self.state.control_packets_tx.fetch_add(1, ORDERING); self.state.control_packets_tx.fetch_add(1, ORDERING);
debug!("tx packet: {:?}", packet); debug!("tx packet: {:?}", packet);
match socket if let Err(e) = socket
.send_to( .send_to(
packet.serialize().unwrap().as_ref(), packet.serialize().unwrap().as_ref(),
SocketAddr::new(dest, CONTROL_PORT), SocketAddr::new(dest, CONTROL_PORT),
) )
.await .await
{ {
Err(e) => { warn!(
warn!( "Error sending packet to {:?}:{} ({:?})",
"Error sending packet to {:?}:{} ({:?})", dest, CONTROL_PORT, e
dest, CONTROL_PORT, e );
);
}
Ok(_) => {}
} }
} }
@@ -639,14 +628,12 @@ impl BfdSession {
}); });
let mut temporary = Some(tx); let mut temporary = Some(tx);
swap(&mut temporary, &mut self.rx_watchdog); swap(&mut temporary, &mut self.rx_watchdog);
if temporary.is_some() { if let Some(old_watchdog) = temporary.as_mut() {
debug!( debug!(
"Updating old watchdog with new detection time ({}ms)", "Updating old watchdog with new detection time ({}ms)",
duration.as_millis() duration.as_millis()
); );
temporary old_watchdog
.as_mut()
.unwrap()
.send(WatchdogReset::Terminate) .send(WatchdogReset::Terminate)
.await .await
.unwrap_or(()) .unwrap_or(())
@@ -662,13 +649,8 @@ impl BfdSession {
// remote system, and bfd.SessionState is Init or Up, the session has // remote system, and bfd.SessionState is Init or Up, the session has
// gone down -- the local system MUST set bfd.SessionState to Down and // gone down -- the local system MUST set bfd.SessionState to Down and
// bfd.LocalDiag to 1 (Control Detection Time Expired). // bfd.LocalDiag to 1 (Control Detection Time Expired).
if self.rx_watchdog.is_some() { if let Some(watchdog) = self.rx_watchdog.as_ref() {
let _ = self let _ = watchdog.send(WatchdogReset::ReceivedPacket).await;
.rx_watchdog
.as_ref()
.unwrap()
.send(WatchdogReset::ReceivedPacket)
.await;
} }
self.state.control_packets_rx.fetch_add(1, ORDERING); self.state.control_packets_rx.fetch_add(1, ORDERING);
let received_state = match p.flags.state() { let received_state = match p.flags.state() {
@@ -774,7 +756,7 @@ impl BfdSession {
self.update_transmit_interval().await; self.update_transmit_interval().await;
// Update the Detection Time as described in section 6.8.4. // Update the Detection Time as described in section 6.8.4.
self.update_detection_time(&p).await; self.update_detection_time(p).await;
// If bfd.SessionState is AdminDown // If bfd.SessionState is AdminDown
// Discard the packet // Discard the packet
@@ -889,16 +871,14 @@ impl BfdSession {
async fn update_detection_time(&mut self, p: &BfdPacket) { async fn update_detection_time(&mut self, p: &BfdPacket) {
let old = self.state.detection_time(); let old = self.state.detection_time();
self.state.set_detection_time(if !self.state.demand_mode() { self.state.set_detection_time(if !self.state.demand_mode() {
(p.flags.detect_mult() as u32 p.flags.detect_mult() as u32
* std::cmp::max(self.state.required_min_rx_interval(), p.desired_min_tx)) * std::cmp::max(self.state.required_min_rx_interval(), p.desired_min_tx)
.into()
} else { } else {
(self.state.detect_mult() as u32 self.state.detect_mult() as u32
* std::cmp::max( * std::cmp::max(
self.state.desired_min_tx_interval(), self.state.desired_min_tx_interval(),
self.state.remote_min_rx_interval(), self.state.remote_min_rx_interval(),
)) )
.into()
}); });
if old != self.state.detection_time() { if old != self.state.detection_time() {
info!( info!(
@@ -1032,7 +1012,7 @@ impl Bfdd {
// If BFD authentication is not in use on a session, all BFD Control packets for the session MUST be sent with a // If BFD authentication is not in use on a session, all BFD Control packets for the session MUST be sent with a
// Time to Live (TTL) or Hop Limit value of 255. // Time to Live (TTL) or Hop Limit value of 255.
set_ttl_or_hop_limit(&control_socket, 255)?; set_ttl_or_hop_limit(&control_socket, 255)?;
let echo_socket = Arc::new(UdpSocket::bind(SocketAddr::new(local.ip(), ECHO_PORT)).await?); let _echo_socket = Arc::new(UdpSocket::bind(SocketAddr::new(local.ip(), ECHO_PORT)).await?);
let event_dispatcher = EventDispatcherHandle::new().await; let event_dispatcher = EventDispatcherHandle::new().await;
event_dispatcher event_dispatcher
@@ -1052,7 +1032,7 @@ impl Bfdd {
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);
while local_discr.0 == 0 || self.sessions_by_discr.get(&local_discr).is_some() { while local_discr.0 == 0 || self.sessions_by_discr.contains_key(&local_discr) {
local_discr = BfdDiscriminator(rng.gen()); local_discr = BfdDiscriminator(rng.gen());
} }
let session = BfdSessionHandle::new( let session = BfdSessionHandle::new(
@@ -1112,7 +1092,6 @@ impl Bfdd {
.unwrap(); .unwrap();
} else { } else {
warn!("Unable to match packet to session with"); warn!("Unable to match packet to session with");
return;
} }
} }
async fn run(mut self) -> io::Result<()> { async fn run(mut self) -> io::Result<()> {
@@ -1187,6 +1166,6 @@ async fn main() -> io::Result<()> {
match Bfdd::new(args).await { match Bfdd::new(args).await {
Ok(bfdd) => bfdd.run().await, Ok(bfdd) => bfdd.run().await,
Err(e) => Err(std::io::Error::new(ErrorKind::Other, e.to_string())), Err(e) => Err(std::io::Error::other(e.to_string())),
} }
} }
+10 -5
View File
@@ -3,12 +3,11 @@ use std::{
ffi::{OsStr, OsString}, ffi::{OsStr, OsString},
net::IpAddr, net::IpAddr,
path::PathBuf, path::PathBuf,
process::ExitStatus,
}; };
use itertools::Itertools; use itertools::Itertools;
use log::{debug, info, warn}; use log::{debug, info, warn};
use nix::unistd::{self, eaccess}; use nix::unistd;
use tokio::{ use tokio::{
fs, process, fs, process,
sync::mpsc, sync::mpsc,
@@ -53,6 +52,12 @@ impl EventMessageSink for ScriptHookSink {
} }
} }
impl Default for ScriptHookSink {
fn default() -> Self {
Self::new()
}
}
impl ScriptHookSink { impl ScriptHookSink {
pub fn new() -> Self { pub fn new() -> Self {
let (tx_channel, rx_channel) = mpsc::channel(16); let (tx_channel, rx_channel) = mpsc::channel(16);
@@ -83,13 +88,13 @@ impl ScriptHookSink {
output.status output.status
); );
if !output.status.success() { if !output.status.success() {
if output.stdout.len() > 0 { if !output.stdout.is_empty() {
warn!( warn!(
" stdout: \n{}", " stdout: \n{}",
str::from_utf8(output.stdout.as_slice()).unwrap() str::from_utf8(output.stdout.as_slice()).unwrap()
); );
} }
if output.stderr.len() > 0 { if !output.stderr.is_empty() {
warn!( warn!(
" stderr: \n{}", " stderr: \n{}",
str::from_utf8(output.stderr.as_slice()).unwrap() str::from_utf8(output.stderr.as_slice()).unwrap()
@@ -133,7 +138,7 @@ impl ScriptHookSink {
.map_or(OsString::new(), |file| file.to_owned()) .map_or(OsString::new(), |file| file.to_owned())
}); });
if scripts.len() > 0 { if !scripts.is_empty() {
// Build context // Build context
let mut args = Vec::<&OsStr>::new(); let mut args = Vec::<&OsStr>::new();
let mut env = Vec::<(&OsStr, &OsStr)>::new(); let mut env = Vec::<(&OsStr, &OsStr)>::new();