From bfd8cecdf59d05f6bc88320316993ab0928baed2 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Wed, 30 Sep 2026 15:34:54 +0700 Subject: [PATCH] feat(measurement): implement ICMP measurement engine --- src/measurement/icmp.rs | 289 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 289 insertions(+) create mode 100644 src/measurement/icmp.rs diff --git a/src/measurement/icmp.rs b/src/measurement/icmp.rs new file mode 100644 index 0000000..8bdbb19 --- /dev/null +++ b/src/measurement/icmp.rs @@ -0,0 +1,289 @@ +use std::{ + collections::HashMap, + io, + net::IpAddr, + time::{Duration, Instant, SystemTime}, +}; + +use flume::Sender; +use icmp_socket2::{ + IcmpSocket, IcmpSocket4, Icmpv4Message, Icmpv4Packet, packet::IcmpPacketBuildError, +}; +use thiserror::Error; + +use crate::{ + measurement::model::{ + MeasurementEvent, MeasurementObservation, MeasurementSource, MeasurementStream, + MeasurementStreamId, OneWayClock, OneWayLatency, SignedDuration, + }, + settings::NetworkSettings, + time::{ClockSample, Realtime}, +}; + +pub struct IcmpBinding { + pub stream: MeasurementStream, + pub interval: Duration, + + next_probe_at: Instant, +} + +#[derive(Debug, Error)] +pub enum IcmpError { + #[error("I/O error")] + Io(#[from] io::Error), + + #[error("failed to build ICMP packet")] + PacketBuild(#[from] IcmpPacketBuildError), + + #[error("invalid ICMP packet: {0}")] + InvalidPacket(String), + + #[error("unsupported measurement source: {0:?}")] + UnsupportedSource(MeasurementSource), + + #[error("measurement observation channel closed")] + ObservationChannelClosed, + + #[error( + "no matching probe found for peer {:?}, source: {:?}, sequence: {}", + .0.0, + .0.1, + .0.2 + )] + NoMatchingProbe(InFlightProbeKey), + + #[error("wrong identifier: expected {expected}, found {found}")] + WrongIdentifier { expected: u16, found: u16 }, +} + +#[derive(Debug, Clone, Copy)] +struct InFlightProbe { + stream: MeasurementStream, + sent_at: Instant, +} + +type InFlightProbeKey = (IpAddr, MeasurementSource, u16); + +pub struct IcmpEngine { + rx_socket: IcmpSocket4, + tx_socket: IcmpSocket4, + identifier: u16, + next_sequence: u16, + + bindings: HashMap, + inflight: HashMap, + + observation_tx: Sender, +} + +impl IcmpEngine { + pub fn new( + #[allow(unused)] settings: NetworkSettings, + observation_tx: Sender, + ) -> Result { + #[allow(unused_mut)] + let mut rx_socket = IcmpSocket4::new()?; + + #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))] + if let Some(device) = &settings.measurement_bind_device { + icmp_socket2::IcmpSocket::bind_device(&mut rx_socket, device)?; + } + + #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))] + if let Some(mark) = settings.measurement_mark { + icmp_socket2::IcmpSocket::set_mark(&mut rx_socket, mark)?; + } + + #[cfg(target_os = "freebsd")] + if let Some(fib) = settings.measurement_fib { + icmp_socket2::IcmpSocket::set_fib(&mut rx_socket, fib)?; + } + + rx_socket.enable_receive_metadata()?; + let tx_socket = rx_socket.try_clone()?; + + Ok(Self { + rx_socket, + tx_socket, + identifier: (std::process::id() & 0xffff) as u16, + next_sequence: 0, + bindings: HashMap::new(), + inflight: HashMap::new(), + observation_tx, + }) + } + + pub fn set_bindings( + &mut self, + bindings: impl IntoIterator, + ) -> Result<(), IcmpError> { + todo!() + } + + fn recv_loop(&mut self) -> Result<(), IcmpError> { + const RECEIVE_TIMEOUT: Duration = Duration::from_millis(50); + const HOUSEKEEPING_INTERVAL: Duration = Duration::from_millis(500); + + let mut next_housekeeping = Instant::now() + HOUSEKEEPING_INTERVAL; + + loop { + match self.rx_socket.rcv_from_with_meta() { + Ok(result) => { + self.handle_reply( + result.packet, + result.peer.as_socket().unwrap().ip(), + result.received_at, + result.kernel_rx_timestamp, + )?; + } + + Err(err) + if matches!( + err.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) => {} + + Err(err) => return Err(err.into()), + } + + let now = Instant::now(); + + if now >= next_housekeeping { + self.expire_inflight(now); + next_housekeeping = now + HOUSEKEEPING_INTERVAL; + } + } + } + + pub fn run(&mut self) -> Result<(), IcmpError> { + todo!() + } + + fn send_due_probes(&mut self, now: Instant) -> Result<(), IcmpError> { + todo!() + } + + fn send_probe( + &mut self, + stream_id: MeasurementStreamId, + now: Instant, + ) -> Result<(), IcmpError> { + todo!() + } + + fn handle_reply( + &mut self, + packet: Icmpv4Packet, + peer: IpAddr, + observed_at: Instant, + kernel_rx_timestamp: Option, + ) -> Result { + match packet.message { + Icmpv4Message::EchoReply { + identifier, + sequence, + .. + } => { + if identifier != self.identifier { + return Err(IcmpError::WrongIdentifier { + expected: self.identifier, + found: identifier, + }); + } + + let key = (peer, MeasurementSource::IcmpEcho, sequence); + + let probe = self + .inflight + .remove(&key) + .ok_or(IcmpError::NoMatchingProbe(key))?; + + Ok(MeasurementObservation { + stream: probe.stream, + started_at: probe.sent_at, + observed_at, + one_way: None, + }) + } + Icmpv4Message::TimestampReply { + identifier, + sequence, + originate, + receive, + transmit, + } => { + const NON_STANDARD_TIMESTAMP: u32 = 1 << 31; + + if identifier != self.identifier { + return Err(IcmpError::WrongIdentifier { + expected: self.identifier, + found: identifier, + }); + } + + let key = (peer, MeasurementSource::IcmpTimestamp, sequence); + + let probe = self + .inflight + .remove(&key) + .ok_or(IcmpError::NoMatchingProbe(key))?; + + if receive & NON_STANDARD_TIMESTAMP != 0 || transmit & NON_STANDARD_TIMESTAMP != 0 { + return Err(IcmpError::InvalidPacket( + "non-standard RFC 792 timestamp".into(), + )); + } + + let now_ms = if let Some(timestamp) = kernel_rx_timestamp { + ClockSample::::from(timestamp).as_time_since_midnight() + } else { + ClockSample::::now().as_time_since_midnight() + }; + + let uplink_ms = timestamp_delta(receive as i64, originate as i64); + let downlink_ms = timestamp_delta(now_ms, transmit as i64); + + Ok(MeasurementObservation { + stream: probe.stream, + started_at: probe.sent_at, + observed_at, + one_way: Some(OneWayLatency { + uplink: SignedDuration::from_millis(uplink_ms), + downlink: SignedDuration::from_millis(downlink_ms), + clock: OneWayClock::Uncalibrated, + }), + }) + } + _ => { + return Err(IcmpError::InvalidPacket( + "unsupported ICMP message type".to_string(), + )); + } + } + } + + fn expire_inflight(&mut self, now: Instant) { + todo!() + } + + fn allocate_sequence(&mut self, peer: IpAddr) -> u16 { + todo!() + } + + fn next_deadline(&self) -> Option { + todo!() + } +} + +const MS_PER_DAY: i64 = 86_400_000; + +fn timestamp_delta(later: i64, earlier: i64) -> i64 { + let delta = later - earlier; + if delta < -MS_PER_DAY / 2 { + delta + MS_PER_DAY + } else if delta > MS_PER_DAY / 2 { + delta - MS_PER_DAY + } else { + delta + } +}