WIP: Add support for setting DSCP/ECN for IPv6 on Windows

crates/trippy-core/Cargo.toml

@@ -31,7 +31,7 @@ nix = { workspace = true, default-features = false, features = ["user", "poll",
 [target.'cfg(windows)'.dependencies]
 paste.workspace = true
 widestring.workspace = true
-windows-sys = { workspace = true, features = ["Win32_Foundation", "Win32_Networking_WinSock", "Win32_System_IO", "Win32_NetworkManagement_IpHelper", "Win32_NetworkManagement_Ndis", "Win32_System_IO", "Win32_System_Threading", "Win32_Security"] }
+windows-sys = { workspace = true, features = ["Win32_Foundation", "Win32_Networking_WinSock", "Win32_System_IO", "Win32_NetworkManagement_IpHelper", "Win32_NetworkManagement_Ndis", "Win32_NetworkManagement_QoS", "Win32_System_IO", "Win32_System_Threading", "Win32_Security"] }
 
 [dev-dependencies]
 anyhow.workspace = true

crates/trippy-core/src/error.rs

@@ -86,6 +86,7 @@ pub enum IoOperation {
     PeerAddr,
     TakeError,
     SetTos,
+    SetTclassV6,
     SetTtl,
     SetReusePort,
     SetHeaderIncluded,
@@ -101,6 +102,10 @@ pub enum IoOperation {
     ConvertSocketAddress,
     SioRoutingInterfaceQuery,
     Startup,
+    QOSCreateHandle,
+    QOSAddSocketToFlow,
+    QOSSetFlow,
+    QOSCloseHandle,
 }
 
 impl Display for IoOperation {
@@ -116,6 +121,7 @@ impl Display for IoOperation {
             Self::PeerAddr => write!(f, "peer addr"),
             Self::TakeError => write!(f, "take error"),
             Self::SetTos => write!(f, "set TOS"),
+            Self::SetTclassV6 => write!(f, "set TCLASS v6"),
             Self::SetTtl => write!(f, "set TTL"),
             Self::SetReusePort => write!(f, "set reuse port"),
             Self::SetHeaderIncluded => write!(f, "set header included"),
@@ -130,6 +136,10 @@ impl Display for IoOperation {
             Self::TcpIcmpErrorInfo => write!(f, "get TCP ICMP error info"),
             Self::ConvertSocketAddress => write!(f, "convert socket address"),
             Self::SioRoutingInterfaceQuery => write!(f, "SIO routing interface query"),
+            Self::QOSCreateHandle => write!(f, "QOS create handle"),
+            Self::QOSAddSocketToFlow => write!(f, "QOS add socket to flow"),
+            Self::QOSSetFlow => write!(f, "QOS set flow"),
+            Self::QOSCloseHandle => write!(f, "QOS close handle"),
             Self::Startup => write!(f, "startup"),
         }
     }

crates/trippy-core/src/net/channel.rs

@@ -69,6 +69,7 @@ impl<S: Socket> Channel<S> {
                 packet_size: config.packet_size,
                 payload_pattern: config.payload_pattern,
                 privilege_mode: config.privilege_mode,
+                tos: config.tos,
                 protocol: config.protocol,
                 icmp_extension_mode: config.icmp_extension_parse_mode,
                 initial_sequence: config.initial_sequence,

crates/trippy-core/src/net/ipv6.rs

@@ -55,6 +55,7 @@ pub struct Ipv6 {
     pub packet_size: PacketSize,
     pub payload_pattern: PayloadPattern,
     pub privilege_mode: PrivilegeMode,
+    pub tos: TypeOfService,
     pub protocol: Protocol,
     pub icmp_extension_mode: IcmpExtensionParseMode,
     pub initial_sequence: Sequence,
@@ -68,6 +69,7 @@ impl Default for Ipv6 {
             packet_size: PacketSize(0),
             payload_pattern: PayloadPattern(0),
             privilege_mode: PrivilegeMode::Privileged,
+            tos: TypeOfService(0),
             protocol: Protocol::Icmp,
             icmp_extension_mode: IcmpExtensionParseMode::Disabled,
             initial_sequence: Sequence(0),
@@ -95,6 +97,7 @@ impl Ipv6 {
             icmp_payload_size(packet_size),
         )?;
         icmp_send_socket.set_unicast_hops_v6(probe.ttl.0)?;
+        icmp_send_socket.set_tclass_v6(u32::from(self.tos.0))?;
         let remote_addr = SocketAddr::new(IpAddr::V6(self.dest_addr), 0);
         icmp_send_socket.send_to(echo_request.packet(), remote_addr)?;
         Ok(())
@@ -149,6 +152,7 @@ impl Ipv6 {
             udp.set_payload(&checksum);
         }
         udp_send_socket.set_unicast_hops_v6(probe.ttl.0)?;
+        udp_send_socket.set_tclass_v6(u32::from(self.tos.0))?;
         // Note that we set the port to be 0 in the remote `SocketAddr` as the target port is
         // encoded in the `UDP` packet.  If we (redundantly) set the target port here then
         // the `send_to` will fail with `EINVAL`.
@@ -168,6 +172,7 @@ impl Ipv6 {
             .or_else(ErrorMapper::in_progress)
             .map_err(|err| ErrorMapper::addr_in_use(err, local_addr))?;
         socket.set_unicast_hops_v6(probe.ttl.0)?;
+        socket.set_tclass_v6(u32::from(self.tos.0))?;
         socket.send_to(payload, remote_addr)?;
         Ok(())
     }
@@ -183,6 +188,7 @@ impl Ipv6 {
             .or_else(ErrorMapper::in_progress)
             .map_err(|err| ErrorMapper::addr_in_use(err, local_addr))?;
         socket.set_unicast_hops_v6(probe.ttl.0)?;
+        socket.set_tclass_v6(u32::from(self.tos.0))?;
         let remote_addr = SocketAddr::new(IpAddr::V6(self.dest_addr), probe.dest_port.0);
         socket
             .connect(remote_addr)
@@ -518,6 +524,11 @@ mod tests {
             .times(1)
             .with(predicate::eq(10))
             .returning(|_| Ok(()));
+        mocket
+            .expect_set_tclass_v6()
+            .times(1)
+            .with(predicate::eq(0))
+            .returning(|_| Ok(()));
         let ipv6 = Ipv6 {
             src_addr,
             dest_addr,
@@ -558,6 +569,11 @@ mod tests {
             .times(1)
             .with(predicate::eq(10))
             .returning(|_| Ok(()));
+        mocket
+            .expect_set_tclass_v6()
+            .times(1)
+            .with(predicate::eq(0))
+            .returning(|_| Ok(()));
         let ipv6 = Ipv6 {
             src_addr,
             dest_addr,
@@ -635,6 +651,11 @@ mod tests {
             .times(1)
             .with(predicate::eq(10))
             .returning(|_| Ok(()));
+        mocket
+            .expect_set_tclass_v6()
+            .times(1)
+            .with(predicate::eq(0))
+            .returning(|_| Ok(()));
 
         let ipv6 = Ipv6 {
             src_addr,
@@ -679,6 +700,11 @@ mod tests {
             .times(1)
             .with(predicate::eq(10))
             .returning(|_| Ok(()));
+        mocket
+            .expect_set_tclass_v6()
+            .times(1)
+            .with(predicate::eq(0))
+            .returning(|_| Ok(()));
 
         let ipv6 = Ipv6 {
             src_addr,
@@ -728,6 +754,11 @@ mod tests {
             .times(1)
             .with(predicate::eq(10))
             .returning(|_| Ok(()));
+        mocket
+            .expect_set_tclass_v6()
+            .times(1)
+            .with(predicate::eq(0))
+            .returning(|_| Ok(()));
 
         let ipv6 = Ipv6 {
             src_addr,
@@ -783,6 +814,11 @@ mod tests {
             .times(1)
             .with(predicate::eq(10))
             .returning(|_| Ok(()));
+        mocket
+            .expect_set_tclass_v6()
+            .times(1)
+            .with(predicate::eq(0))
+            .returning(|_| Ok(()));
 
         let ipv6 = Ipv6 {
             src_addr,
@@ -829,6 +865,12 @@ mod tests {
                 .with(predicate::eq(expected_set_unicast_hops_v6))
                 .returning(|_| Ok(()));
 
+            mocket
+                .expect_set_tclass_v6()
+                .times(1)
+                .with(predicate::eq(0))
+                .returning(|_| Ok(()));
+
             mocket
                 .expect_send_to()
                 .with(
@@ -886,6 +928,12 @@ mod tests {
                 .with(predicate::eq(expected_set_unicast_hops_v6))
                 .returning(|_| Ok(()));
 
+            mocket
+                .expect_set_tclass_v6()
+                .times(1)
+                .with(predicate::eq(0))
+                .returning(|_| Ok(()));
+
             mocket
                 .expect_send_to()
                 .with(
@@ -984,6 +1032,12 @@ mod tests {
                 .with(predicate::eq(expected_set_unicast_hops_v6))
                 .returning(|_| Ok(()));
 
+            mocket
+                .expect_set_tclass_v6()
+                .times(1)
+                .with(predicate::eq(0))
+                .returning(|_| Ok(()));
+
             mocket
                 .expect_connect()
                 .with(predicate::eq(expected_connect_addr))

crates/trippy-core/src/net/platform/unix.rs

@@ -342,6 +342,12 @@ mod socket {
                 .map_err(|err| IoError::Other(err, IoOperation::SetTos))
         }
         #[instrument(skip(self), level = "trace")]
+        fn set_tclass_v6(&mut self, tclass: u32) -> IoResult<()> {
+            self.inner
+                .set_tclass_v6(tclass)
+                .map_err(|err| IoError::Other(err, IoOperation::SetTclassV6))
+        }
+        #[instrument(skip(self), level = "trace")]
         fn set_ttl(&mut self, ttl: u32) -> IoResult<()> {
             self.inner
                 .set_ttl(ttl)

crates/trippy-core/src/net/platform/windows.rs

@@ -14,14 +14,17 @@ use std::os::windows::prelude::AsRawSocket;
 use std::ptr::{addr_of, addr_of_mut, null_mut};
 use std::time::Duration;
 use tracing::instrument;
-use windows_sys::Win32::Foundation::{WAIT_FAILED, WAIT_TIMEOUT};
+use windows_sys::Win32::Foundation::{HANDLE, WAIT_FAILED, WAIT_TIMEOUT};
+use windows_sys::Win32::NetworkManagement::QoS::{
+    QOSSetOutgoingDSCPValue, QOSTrafficTypeBestEffort, QOS_VERSION,
+};
 use windows_sys::Win32::Networking::WinSock::{
     AF_INET, AF_INET6, FD_CONNECT, FD_WRITE, ICMP_ERROR_INFO, IN6_ADDR, IN6_ADDR_0, IN_ADDR,
     IN_ADDR_0, IPPROTO_RAW, IPPROTO_TCP, SIO_ROUTING_INTERFACE_QUERY, SOCKADDR_IN, SOCKADDR_IN6,
-    SOCKADDR_IN6_0, SOCKADDR_STORAGE, SOCKET_ERROR, SOL_SOCKET, SO_ERROR, SO_PORT_SCALABILITY,
-    SO_REUSE_UNICASTPORT, TCP_FAIL_CONNECT_ON_ICMP_ERROR, TCP_ICMP_ERROR_INFO, WSABUF, WSADATA,
-    WSAEADDRNOTAVAIL, WSAECONNREFUSED, WSAEHOSTUNREACH, WSAEINPROGRESS, WSAENETUNREACH, WSAENOBUFS,
-    WSA_IO_INCOMPLETE, WSA_IO_PENDING,
+    SOCKADDR_IN6_0, SOCKADDR_STORAGE, SOCKET, SOCKET_ERROR, SOL_SOCKET, SO_ERROR,
+    SO_PORT_SCALABILITY, SO_REUSE_UNICASTPORT, TCP_FAIL_CONNECT_ON_ICMP_ERROR, TCP_ICMP_ERROR_INFO,
+    WSABUF, WSADATA, WSAEADDRNOTAVAIL, WSAECONNREFUSED, WSAEHOSTUNREACH, WSAEINPROGRESS,
+    WSAENETUNREACH, WSAENOBUFS, WSA_IO_INCOMPLETE, WSA_IO_PENDING,
 };
 use windows_sys::Win32::System::IO::OVERLAPPED;
 
@@ -62,6 +65,21 @@ macro_rules! syscall_threading {
     }};
 }
 
+/// Execute a `Win32::NetworkManagement::Qos` syscall.
+///
+/// If the syscall fails then the last OS error is returned.
+macro_rules! syscall_qos {
+    ($fn:ident ( $($arg:expr),* $(,)* )) => {{
+        #[allow(unsafe_code)]
+        let res = unsafe { windows_sys::Win32::NetworkManagement::QoS::$fn($($arg, )*) };
+        if res == 0 {
+            Err(StdIoError::last_os_error())
+        } else {
+            Ok(res)
+        }
+    }};
+}
+
 pub struct PlatformImpl;
 
 impl Platform for PlatformImpl {
@@ -415,6 +433,40 @@ impl Socket for SocketImpl {
             .map_err(|err| IoError::Other(err, IoOperation::SetTos))
     }
 
+    fn set_tclass_v6(&mut self, tclass: u32) -> IoResult<()> {
+        let dscp: u32 = (tclass >> 2) & 0x3F;
+        let ver = QOS_VERSION {
+            MajorVersion: 1,
+            MinorVersion: 0,
+        };
+        let mut qos: HANDLE = 0;
+        syscall_qos!(QOSCreateHandle(&ver, &mut qos))
+            .map_err(|err| IoError::Other(err, IoOperation::QOSCreateHandle))?;
+        let mut flow: u32 = 0;
+        syscall_qos!(QOSAddSocketToFlow(
+            qos,
+            self.inner.as_raw_socket() as SOCKET,
+            std::ptr::null(),
+            QOSTrafficTypeBestEffort,
+            0,
+            &mut flow
+        ))
+        .map_err(|err| IoError::Other(err, IoOperation::QOSAddSocketToFlow))?;
+        syscall_qos!(QOSSetFlow(
+            qos,
+            flow,
+            QOSSetOutgoingDSCPValue,
+            size_of::<u32>() as u32,
+            addr_of!(dscp).cast(),
+            0,
+            null_mut()
+        ))
+        .map_err(|err| IoError::Other(err, IoOperation::QOSSetFlow))?;
+        syscall_qos!(QOSCloseHandle(qos))
+            .map_err(|err| IoError::Other(err, IoOperation::QOSCloseHandle))?;
+        Ok(())
+    }
+
     #[instrument(skip(self), level = "trace")]
     fn set_ttl(&mut self, ttl: u32) -> IoResult<()> {
         self.inner

crates/trippy-core/src/net/socket.rs

@@ -29,6 +29,7 @@ where
     fn new_udp_dgram_socket_ipv6() -> Result<Self>;
     fn bind(&mut self, address: SocketAddr) -> Result<()>;
     fn set_tos(&mut self, tos: u32) -> Result<()>;
+    fn set_tclass_v6(&mut self, tclass: u32) -> Result<()>;
     fn set_ttl(&mut self, ttl: u32) -> Result<()>;
     fn set_reuse_port(&mut self, reuse: bool) -> Result<()>;
     fn set_header_included(&mut self, included: bool) -> Result<()>;