feat: multi-session support in reflector

Author: asmie

README.md

@@ -15,7 +15,7 @@ stamp-suite is a Rust implementation of the Simple Two-Way Active Measurement Pr
 
 - Full RFC 8762 compliance (unauthenticated and authenticated modes)
 - HMAC authentication support
-- Stateful reflector mode (RFC 8972)
+- Stateful reflector mode with per-client session tracking (RFC 8972 Section 4)
 - Support for both NTP and PTP timestamp formats
 - Real TTL/Hop Limit capture on all platforms
 - Async I/O using Tokio
@@ -68,6 +68,13 @@ stamp-suite -i --local-addr 192.168.1.100 --local-port 8620
 
 # With verbose per-packet statistics
 stamp-suite -i -R
+
+# Stateful reflector mode (RFC 8972) - maintains independent sequence
+# counters per client, allowing detection of reflector-side packet loss
+stamp-suite -i --stateful-reflector
+
+# Stateful mode with custom session timeout (default: 300 seconds)
+stamp-suite -i --stateful-reflector --session-timeout 600
 ```
 
 ### Session-Sender (Client)
@@ -101,9 +108,11 @@ Options:
   -L, --timeout <TIMEOUT>          Timeout for lost packets in seconds [default: 5]
   -4                               Force IPv4 addresses
   -6                               Force IPv6 addresses
-  -A, --auth-mode <AUTH_MODE>      Work mode: A=auth, E=encrypted, O=open [default: AEO]
+  -A, --auth-mode <AUTH_MODE>      Work mode: A=authenticated, O=open [default: O]
   -R                               Print individual statistics for each packet
   -i, --is-reflector               Run as Session Reflector instead of Sender
+      --stateful-reflector         Enable stateful reflector mode (RFC 8972 Section 4)
+      --session-timeout <SECONDS>  Session timeout for stateful mode [default: 300]
   -h, --help                       Print help
   -V, --version                    Print version
 ```
@@ -152,17 +161,15 @@ The project is functional for STAMP measurements with the following features:
 
 - Full RFC 8762 compliance (unauthenticated and authenticated modes)
 - HMAC authentication support
-- Stateful reflector mode (RFC 8972)
+- Stateful reflector mode with per-client session tracking (RFC 8972 Section 4)
 - Real TTL capture on all major platforms
 
 Current limitations:
 
-- Single session per instance (no multi-session support yet)
 - No STAMP-MIB support
 
 ### Roadmap
 
-- [ ] Multi-session support for reflector
 - [ ] DSCP/ECN field handling
 - [ ] Enhanced statistics and reporting
 

src/configuration.rs

@@ -82,11 +82,16 @@ pub struct Configuration {
     #[clap(long)]
     pub strict_packets: bool,
 
-    /// Use stateful reflector mode with independent sequence counter (RFC 8972).
-    /// When enabled, the reflector maintains its own sequence counter instead of
-    /// echoing the sender's sequence number.
+    /// Enable stateful reflector mode per RFC 8972 Section 4. The reflector maintains
+    /// independent sequence counters for each client (IP:port) instead of echoing
+    /// the sender's sequence number, allowing clients to detect reflector-side packet loss.
     #[clap(long)]
     pub stateful_reflector: bool,
+
+    /// Session timeout in seconds for stateful reflector mode. Sessions inactive for
+    /// this duration may be cleaned up. Default: 300 (5 minutes). Set to 0 to disable.
+    #[clap(long, default_value_t = 300)]
+    pub session_timeout: u64,
 }
 
 impl Configuration {
@@ -102,6 +107,11 @@ impl Configuration {
         }
 
         // Validate auth mode - only A (authenticated) and O (open) are valid per RFC 8762
+        if self.auth_mode.is_empty() {
+            return Err(ConfigurationError::InvalidConfiguration(
+                "Auth mode cannot be empty. Valid options: A (authenticated), O (open)".to_string(),
+            ));
+        }
         for c in self.auth_mode.chars() {
             if c != 'A' && c != 'O' {
                 return Err(ConfigurationError::InvalidConfiguration(format!(
@@ -165,7 +175,7 @@ mod tests {
             "--timeout",
             "5",
             "--auth-mode",
-            "AEO",
+            "AO",
             "--is-reflector",
         ];
         let conf = Configuration::parse_from(args);
@@ -177,8 +187,9 @@ mod tests {
         assert_eq!(conf.send_delay, 1000);
         assert_eq!(conf.count, 1000);
         assert_eq!(conf.timeout, 5);
-        assert_eq!(conf.auth_mode, "AEO");
+        assert_eq!(conf.auth_mode, "AO");
         assert!(conf.is_reflector);
+        assert!(conf.validate().is_ok());
     }
 
     #[test]
@@ -331,9 +342,8 @@ mod tests {
     fn test_auth_mode_empty() {
         let args = vec!["test", "--auth-mode", ""];
         let conf = Configuration::parse_from(args);
-        assert!(conf.validate().is_ok()); // Empty is valid (no modes enabled)
-        assert!(!is_auth(&conf.auth_mode));
-        assert!(!is_open(&conf.auth_mode));
+        // Empty auth mode is invalid - must specify at least one mode
+        assert!(conf.validate().is_err());
     }
 
     #[test]
@@ -447,4 +457,37 @@ mod tests {
 
         assert!(!conf.stateful_reflector);
     }
+
+    #[test]
+    fn test_session_timeout_default() {
+        let args = vec!["test"];
+        let conf = Configuration::parse_from(args);
+
+        assert_eq!(conf.session_timeout, 300);
+    }
+
+    #[test]
+    fn test_session_timeout_custom() {
+        let args = vec!["test", "--session-timeout", "600"];
+        let conf = Configuration::parse_from(args);
+
+        assert_eq!(conf.session_timeout, 600);
+    }
+
+    #[test]
+    fn test_session_timeout_zero_disables() {
+        let args = vec!["test", "--session-timeout", "0"];
+        let conf = Configuration::parse_from(args);
+
+        assert_eq!(conf.session_timeout, 0);
+    }
+
+    #[test]
+    fn test_stateful_reflector_with_timeout() {
+        let args = vec!["test", "--stateful-reflector", "--session-timeout", "120"];
+        let conf = Configuration::parse_from(args);
+
+        assert!(conf.stateful_reflector);
+        assert_eq!(conf.session_timeout, 120);
+    }
 }

src/receiver/mod.rs

@@ -99,11 +99,12 @@ pub const AUTH_BASE_SIZE: usize = 112;
 
 /// Assembles an unauthenticated reflected packet with symmetric size (RFC 8762 Section 4.3).
 ///
-/// Preserves the original packet length by copying extra bytes beyond the base 44 bytes.
+/// Preserves the original packet length by padding with zeros beyond the base 44 bytes.
+/// Per RFC 8762 Section 4.2.1, extra octets SHOULD be filled with zeros.
 ///
 /// # Arguments
 /// * `packet` - The received unauthenticated test packet
-/// * `original_data` - The original received packet data
+/// * `original_data` - The original received packet data (used only for length)
 /// * `cs` - Clock format to use for timestamps
 /// * `rcvt` - Receive timestamp when the packet was received
 /// * `ttl` - TTL/Hop Limit value from the received packet's IP header
@@ -128,9 +129,9 @@ pub fn assemble_unauth_answer_symmetric(
     );
     let mut response = base.to_bytes().to_vec();
 
-    // Copy extra bytes beyond base packet (RFC 8762 Section 4.3)
+    // Pad with zeros to match original length (RFC 8762 Section 4.2.1)
     if original_data.len() > UNAUTH_BASE_SIZE {
-        response.extend_from_slice(&original_data[UNAUTH_BASE_SIZE..]);
+        response.resize(original_data.len(), 0);
     }
 
     response
@@ -184,11 +185,12 @@ pub fn assemble_auth_answer(
 
 /// Assembles an authenticated reflected packet with symmetric size (RFC 8762 Section 4.3).
 ///
-/// Preserves the original packet length by copying extra bytes beyond the base 112 bytes.
+/// Preserves the original packet length by padding with zeros beyond the base 112 bytes.
+/// Per RFC 8762 Section 4.2.1, extra octets SHOULD be filled with zeros.
 ///
 /// # Arguments
 /// * `packet` - The received authenticated test packet
-/// * `original_data` - The original received packet data
+/// * `original_data` - The original received packet data (used only for length)
 /// * `cs` - Clock format to use for timestamps
 /// * `rcvt` - Receive timestamp when the packet was received
 /// * `ttl` - TTL/Hop Limit value from the received packet's IP header
@@ -217,9 +219,9 @@ pub fn assemble_auth_answer_symmetric(
     );
     let mut response = base.to_bytes().to_vec();
 
-    // Copy extra bytes beyond base packet (RFC 8762 Section 4.3)
+    // Pad with zeros to match original length (RFC 8762 Section 4.2.1)
     if original_data.len() > AUTH_BASE_SIZE {
-        response.extend_from_slice(&original_data[AUTH_BASE_SIZE..]);
+        response.resize(original_data.len(), 0);
     }
 
     response
@@ -511,8 +513,8 @@ mod tests {
 
         // Response should be 48 bytes (44 base + 4 extra)
         assert_eq!(response.len(), 48);
-        // Extra bytes should be preserved at the end
-        assert_eq!(&response[44..], &[0xAA, 0xBB, 0xCC, 0xDD]);
+        // Extra bytes should be zeros per RFC 8762 Section 4.2.1
+        assert_eq!(&response[44..], &[0x00, 0x00, 0x00, 0x00]);
     }
 
     #[test]
@@ -545,8 +547,8 @@ mod tests {
 
         // Response should be 117 bytes (112 base + 5 extra)
         assert_eq!(response.len(), 117);
-        // Extra bytes should be preserved at the end
-        assert_eq!(&response[112..], &[0x11, 0x22, 0x33, 0x44, 0x55]);
+        // Extra bytes should be zeros per RFC 8762 Section 4.2.1
+        assert_eq!(&response[112..], &[0x00, 0x00, 0x00, 0x00, 0x00]);
     }
 
     #[test]

src/receiver/nix.rs

@@ -2,7 +2,7 @@
 //!
 //! Preferred on Linux systems. No special privileges required for regular UDP sockets.
 
-use std::{io::IoSliceMut, net::SocketAddr, os::fd::AsRawFd};
+use std::{io::IoSliceMut, net::SocketAddr, os::fd::AsRawFd, sync::Arc};
 
 use nix::{
     libc,
@@ -15,7 +15,7 @@ use crate::{
     crypto::HmacKey,
     error_estimate::ErrorEstimate,
     packets::{PacketAuthenticated, PacketUnauthenticated},
-    session::Session,
+    session::SessionManager,
     time::generate_timestamp,
 };
 
@@ -109,10 +109,15 @@ pub async fn run_receiver(conf: &Configuration) {
         log::info!("HMAC authentication enabled");
     }
 
-    // Create session for stateful reflector mode (RFC 8972)
-    let reflector_session = if conf.stateful_reflector {
+    // Create session manager for stateful reflector mode (RFC 8972)
+    let session_manager: Option<Arc<SessionManager>> = if conf.stateful_reflector {
+        let timeout = if conf.session_timeout > 0 {
+            Some(std::time::Duration::from_secs(conf.session_timeout))
+        } else {
+            None
+        };
         log::info!("Stateful reflector mode enabled (RFC 8972)");
-        Some(Session::new(0))
+        Some(Arc::new(SessionManager::new(timeout)))
     } else {
         None
     };
@@ -143,7 +148,7 @@ pub async fn run_receiver(conf: &Configuration) {
         ) {
             Ok(msg) => {
                 let len = msg.bytes;
-                let src_addr = msg.address;
+                let src_storage = msg.address;
 
                 // Extract TTL from control messages
                 let ttl = match extract_ttl_from_cmsgs(&msg) {
@@ -153,6 +158,25 @@ pub async fn run_receiver(conf: &Configuration) {
                         continue;
                     }
                 };
+
+                // Convert source address for session lookup and response
+                let src_addr: SocketAddr = match src_storage {
+                    Some(ref src) => {
+                        if let Some(v4) = src.as_sockaddr_in() {
+                            std::net::SocketAddrV4::new(v4.ip(), v4.port()).into()
+                        } else if let Some(v6) = src.as_sockaddr_in6() {
+                            std::net::SocketAddrV6::new(v6.ip(), v6.port(), 0, 0).into()
+                        } else {
+                            eprintln!("Unknown source address type");
+                            continue;
+                        }
+                    }
+                    None => {
+                        eprintln!("No source address available");
+                        continue;
+                    }
+                };
+
                 let rcvt = generate_timestamp(conf.clock_source);
 
                 let response = if use_auth {
@@ -179,9 +203,9 @@ pub async fn run_receiver(conf: &Configuration) {
                             }
 
                             // Generate reflector sequence number only after successful validation
-                            let reflector_seq = reflector_session
+                            let reflector_seq = session_manager
                                 .as_ref()
-                                .map(|s| s.generate_sequence_number());
+                                .map(|mgr| mgr.generate_sequence_number(src_addr));
 
                             // Use symmetric assembly to preserve original packet length (RFC 8762 Section 4.3)
                             Some(assemble_auth_answer_symmetric(
@@ -209,9 +233,9 @@ pub async fn run_receiver(conf: &Configuration) {
                     match packet_result {
                         Ok(packet) => {
                             // Generate reflector sequence number only after successful validation
-                            let reflector_seq = reflector_session
+                            let reflector_seq = session_manager
                                 .as_ref()
-                                .map(|s| s.generate_sequence_number());
+                                .map(|mgr| mgr.generate_sequence_number(src_addr));
 
                             // Use symmetric assembly to preserve original packet length (RFC 8762 Section 4.3)
                             Some(assemble_unauth_answer_symmetric(
@@ -232,20 +256,8 @@ pub async fn run_receiver(conf: &Configuration) {
                 };
 
                 if let Some(response_buf) = response {
-                    if let Some(src) = src_addr {
-                        // Convert SockaddrStorage back to SocketAddr
-                        let dest: SocketAddr = if let Some(v4) = src.as_sockaddr_in() {
-                            std::net::SocketAddrV4::new(v4.ip(), v4.port()).into()
-                        } else if let Some(v6) = src.as_sockaddr_in6() {
-                            std::net::SocketAddrV6::new(v6.ip(), v6.port(), 0, 0).into()
-                        } else {
-                            eprintln!("Unknown source address type");
-                            continue;
-                        };
-
-                        if let Err(e) = tokio_socket.send_to(&response_buf, dest).await {
-                            eprintln!("Failed to send response: {}", e);
-                        }
+                    if let Err(e) = tokio_socket.send_to(&response_buf, src_addr).await {
+                        eprintln!("Failed to send response: {}", e);
                     }
                 }
             }