Reject stale MMP receiver reports

Ignore duplicate or counter-regressed ReceiverReports before updating
RTT, loss, goodput, or ETX, so a delayed or reordered report can no
longer poison link metrics. Compute the RTT-from-echo sample with
checked timestamp arithmetic and reject zero, negative, or out-of-range
results instead of risking wrap or underflow on untrusted wire values.

On the sender side, when receiver dwell time overflows the u16 wire
field, suppress the timestamp echo (send 0) and saturate dwell to
u16::MAX rather than truncating, so a bogus small RTT cannot be formed.

Adds duplicate, out-of-order, wrapped-add, and future-dated (checked_sub)
sample tests, asserts loss and goodput stay unchanged on a dropped
duplicate, and covers the dwell-overflow echo suppression. Documents the
behavior in the MMP design note and CHANGELOG.

Co-authored-by: Johnathan Corgan <johnathan@corganlabs.com>

CHANGELOG.md

@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- MMP sender metrics now ignore duplicate or regressed receiver reports
+  before updating RTT, loss, goodput, or ETX. Receiver reports also
+  suppress timestamp echo when dwell time overflows, so stale reports
+  cannot inflate SRTT.
+
 ### Added
 
 - `pool_inbound` and `pool_outbound` counters on the TCP and Tor

docs/design/fips-mmp.md

@@ -100,6 +100,11 @@ inter-frame processing delays inflate spin bit RTT measurements
 unpredictably. Timestamp-echo from ReceiverReports (with dwell-time
 compensation) is the sole SRTT source.
 
+Duplicate or regressed ReceiverReports are ignored before any RTT, loss,
+goodput, or ETX update. If receiver-side dwell time exceeds the wire
+field, the report keeps its counters but sends a zero timestamp echo so
+the sender cannot form an invalid RTT sample.
+
 The spin bit lives in the link-layer FMP inner header, so this
 mechanism applies to link-layer MMP only. Session-layer MMP carries
 its spin bit in the FSP encrypted inner header but uses it the same

src/mmp/metrics.rs

@@ -118,25 +118,64 @@ impl MmpMetrics {
     ) -> bool {
         let had_srtt = self.srtt.initialized();
 
+        if self.has_prev_rr {
+            let counters_regressed = rr.highest_counter < self.prev_rr_highest_counter
+                || rr.cumulative_packets_recv < self.prev_rr_cum_packets
+                || rr.cumulative_bytes_recv < self.prev_rr_cum_bytes
+                || rr.ecn_ce_count < self.prev_rr_ecn_ce
+                || rr.cumulative_reorder_count < self.prev_rr_reorder;
+            let duplicate_counters = rr.highest_counter == self.prev_rr_highest_counter
+                && rr.cumulative_packets_recv == self.prev_rr_cum_packets
+                && rr.cumulative_bytes_recv == self.prev_rr_cum_bytes
+                && rr.ecn_ce_count == self.prev_rr_ecn_ce
+                && rr.cumulative_reorder_count == self.prev_rr_reorder;
+            // Safe to drop: reports are only built after interval data, so
+            // a fresh report always advances at least one cumulative counter.
+            if counters_regressed || duplicate_counters {
+                trace!(
+                    highest_counter = rr.highest_counter,
+                    prev_highest_counter = self.prev_rr_highest_counter,
+                    cumulative_packets_recv = rr.cumulative_packets_recv,
+                    prev_cumulative_packets_recv = self.prev_rr_cum_packets,
+                    cumulative_bytes_recv = rr.cumulative_bytes_recv,
+                    prev_cumulative_bytes_recv = self.prev_rr_cum_bytes,
+                    "Ignoring stale MMP ReceiverReport"
+                );
+                return false;
+            }
+        }
+
         // --- RTT from timestamp echo ---
         // RTT = now - echoed_timestamp - dwell_time
         if rr.timestamp_echo > 0 {
             let echo_ms = rr.timestamp_echo;
-            let dwell_ms = rr.dwell_time as u32;
-            // Guard against timestamp wrap or bogus values
-            if our_timestamp_ms > echo_ms + dwell_ms {
-                let rtt_ms = our_timestamp_ms - echo_ms - dwell_ms;
-                let rtt_us = (rtt_ms as i64) * 1000;
-                trace!(
-                    our_ts = our_timestamp_ms,
-                    echo = echo_ms,
-                    dwell = dwell_ms,
-                    rtt_ms = rtt_ms,
-                    srtt_ms = self.srtt.srtt_us() as f64 / 1000.0,
-                    "RTT sample from timestamp echo"
-                );
-                self.srtt.update(rtt_us);
-                self.rtt_trend.update(rtt_us as f64);
+            let dwell_ms = u32::from(rr.dwell_time);
+            let rtt_sample_ms = echo_ms
+                .checked_add(dwell_ms)
+                .and_then(|send_done_ms| our_timestamp_ms.checked_sub(send_done_ms));
+
+            match rtt_sample_ms {
+                Some(rtt_ms) if rtt_ms > 0 => {
+                    let rtt_us = (rtt_ms as i64) * 1000;
+                    trace!(
+                        our_ts = our_timestamp_ms,
+                        echo = echo_ms,
+                        dwell = dwell_ms,
+                        rtt_ms = rtt_ms,
+                        srtt_ms = self.srtt.srtt_us() as f64 / 1000.0,
+                        "RTT sample from timestamp echo"
+                    );
+                    self.srtt.update(rtt_us);
+                    self.rtt_trend.update(rtt_us as f64);
+                }
+                _ => {
+                    trace!(
+                        our_ts = our_timestamp_ms,
+                        echo = echo_ms,
+                        dwell = dwell_ms,
+                        "Ignoring invalid MMP RTT sample"
+                    );
+                }
             }
         }
 
@@ -318,6 +357,70 @@ mod tests {
         assert!((srtt_ms - 45.0).abs() < 1.0, "srtt={srtt_ms}, expected ~45");
     }
 
+    #[test]
+    fn test_ignores_duplicate_receiver_report_after_valid_sample() {
+        let mut m = MmpMetrics::new();
+        let t0 = Instant::now();
+
+        let rr1 = make_rr(10, 10, 5_000, 1_000, 5, 0);
+        m.process_receiver_report(&rr1, 1_050, t0);
+
+        let rr2 = make_rr(20, 18, 14_000, 1_100, 5, 0);
+        m.process_receiver_report(&rr2, 1_150, t0 + Duration::from_secs(1));
+        let baseline_srtt_ms = m.srtt_ms().unwrap();
+        let baseline_loss = m.loss_rate();
+        let baseline_goodput = m.goodput_bps();
+
+        assert!(baseline_loss > 0.0);
+        assert!(baseline_goodput > 0.0);
+
+        // A duplicate of the same counters arriving later would be a 4.895s
+        // RTT sample if accepted. It is stale and must not move metrics.
+        m.process_receiver_report(&rr2, 6_000, t0 + Duration::from_secs(5));
+
+        assert_eq!(m.srtt_ms().unwrap(), baseline_srtt_ms);
+        assert_eq!(m.loss_rate(), baseline_loss);
+        assert_eq!(m.goodput_bps(), baseline_goodput);
+    }
+
+    #[test]
+    fn test_ignores_out_of_order_receiver_report_after_valid_sample() {
+        let mut m = MmpMetrics::new();
+        let now = Instant::now();
+
+        let valid_rr = make_rr(20, 20, 10000, 1000, 5, 0);
+        m.process_receiver_report(&valid_rr, 1050, now);
+        let baseline_srtt_ms = m.srtt_ms().unwrap();
+
+        let old_rr = make_rr(10, 10, 5000, 1000, 0, 0);
+        m.process_receiver_report(&old_rr, 6000, now + Duration::from_secs(5));
+
+        let srtt_ms = m.srtt_ms().unwrap();
+        assert_eq!(srtt_ms, baseline_srtt_ms);
+    }
+
+    #[test]
+    fn test_ignores_wrapped_rtt_sample() {
+        let mut m = MmpMetrics::new();
+        let now = Instant::now();
+
+        let wrapped_rr = make_rr(10, 10, 5000, u32::MAX - 10, 20, 0);
+        m.process_receiver_report(&wrapped_rr, 15, now);
+
+        assert!(m.srtt_ms().is_none());
+    }
+
+    #[test]
+    fn test_ignores_future_rtt_sample() {
+        let mut m = MmpMetrics::new();
+        let now = Instant::now();
+
+        let future_rr = make_rr(10, 10, 5_000, 2_000, 5, 0);
+        m.process_receiver_report(&future_rr, 1_000, now);
+
+        assert!(m.srtt_ms().is_none());
+    }
+
     #[test]
     fn test_loss_rate_computation() {
         let mut m = MmpMetrics::new();

src/mmp/receiver.rs

@@ -323,11 +323,20 @@ impl ReceiverState {
             return None;
         }
 
-        // Dwell time: ms between last frame reception and report generation
-        let dwell_time = self
+        // Dwell time: ms between last frame reception and report generation.
+        // If it no longer fits on the wire, the timestamp echo cannot produce
+        // a valid RTT sample. Preserve the counters but suppress the echo.
+        let (timestamp_echo, dwell_time) = self
             .last_recv_time
-            .map(|t| now.duration_since(t).as_millis() as u16)
-            .unwrap_or(0);
+            .map(|t| {
+                let dwell_ms = now.duration_since(t).as_millis();
+                if dwell_ms > u128::from(u16::MAX) {
+                    (0, u16::MAX)
+                } else {
+                    (self.last_sender_timestamp, dwell_ms as u16)
+                }
+            })
+            .unwrap_or((0, 0));
 
         let (burst_count, max_burst, mean_burst) = self.gap_tracker.take_interval_stats();
 
@@ -335,7 +344,7 @@ impl ReceiverState {
             highest_counter: self.highest_counter,
             cumulative_packets_recv: self.cumulative_packets_recv,
             cumulative_bytes_recv: self.cumulative_bytes_recv,
-            timestamp_echo: self.last_sender_timestamp,
+            timestamp_echo,
             dwell_time,
             max_burst_loss: max_burst,
             mean_burst_loss: mean_burst,
@@ -505,6 +514,21 @@ mod tests {
         assert_eq!(report.interval_bytes_recv, 1100);
     }
 
+    #[test]
+    fn test_build_report_suppresses_rtt_echo_when_dwell_overflows() {
+        let mut r = ReceiverState::new(32);
+        let t0 = Instant::now();
+        r.record_recv(1, 100, 500, false, t0);
+
+        let report = r
+            .build_report(t0 + Duration::from_millis(u64::from(u16::MAX) + 1))
+            .unwrap();
+
+        assert_eq!(report.timestamp_echo, 0);
+        assert_eq!(report.dwell_time, u16::MAX);
+        assert_eq!(report.cumulative_packets_recv, 1);
+    }
+
     #[test]
     fn test_build_report_resets_interval() {
         let mut r = ReceiverState::new(32);