feat(socket): add Linux SO_MARK (fwmark) support for underlay sockets (#2288)

Adds a Linux-only socket_mark u32 config flag (CLI: --socket-mark, env:
ET_SOCKET_MARK, TOML/proto: flags.socket_mark, 0 = disabled) that is
applied as SO_MARK to every outbound underlay socket EasyTier creates:
TCP, UDP, QUIC, WebSocket, WireGuard connectors and listeners, plus the
FakeTCP decoy socket. Lets the host policy-route or filter EasyTier
underlay traffic with 'ip rule fwmark ...' or iptables -m mark.

Plumbing mirrors the existing bind_device pattern:
- FlagsInConfig.socket_mark (proto) + default 0 in gen_default_flags
- bind() builder gets a socket_mark arg; setup_socket2_ext calls
  apply_socket_mark which is a no-op for mark=0 and on non-Linux
- TunnelConnector trait gets set_socket_mark(u32) default-no-op method
- IP-based connectors override; create_listener_by_url and the connector
  factory pass mark from global_ctx flags
- QUIC threads mark through QuicEndpointManager::{server,connect}
- WebSocket/FakeTCP/TCP default-bind bypass paths apply mark via
  socket2::SockRef::from(&tokio_socket)
- ForeignNetworkEntry propagates parent socket_mark into its derived ctx

Includes a Linux smoke test plus a CAP_NET_ADMIN-gated test that does a
getsockopt(SO_MARK) round-trip to confirm the kernel applied the value.

SO_MARK requires CAP_NET_ADMIN; ignored silently on non-Linux. FakeTCP's
TUN-written segments are not covered (kernel doesn't tag raw TUN
writes); operators relying on fwmark for FakeTCP must apply an iptables
rule on the FakeTCP TUN device separately.

Co-authored-by: Claude <noreply@anthropic.com>

easytier/locales/app.yml

@@ -205,6 +205,9 @@ core_clap:
   bind_device:
     en: "bind the connector socket to physical devices to avoid routing issues. e.g.: subnet proxy segment conflicts with a node's segment, after binding the physical device, it can communicate with the node normally."
     zh-CN: "将连接器的套接字绑定到物理设备以避免路由问题。比如子网代理网段与某节点的网段冲突，绑定物理设备后可以与该节点正常通信。"
+  socket_mark:
+    en: "Linux only: set SO_MARK (fwmark) on EasyTier's underlay sockets (TCP, UDP, QUIC, WebSocket, WireGuard, and the FakeTCP decoy socket) so the host can policy-route or filter them with 'ip rule fwmark ...', nftables ('meta mark'), or iptables ('-m mark'). Any value is applied verbatim (0 is a valid mark); omit the flag to leave SO_MARK untouched. Requires CAP_NET_ADMIN. Note: FakeTCP payload travels via raw TUN writes which the kernel does not tag — mark those separately on the TUN device if needed."
+    zh-CN: "仅 Linux: 在 EasyTier 的底层套接字 (TCP、UDP、QUIC、WebSocket、WireGuard 以及 FakeTCP 诱饵套接字) 上设置 SO_MARK (fwmark)，使主机能用 'ip rule fwmark ...'、nftables ('meta mark') 或 iptables ('-m mark') 策略路由/过滤这些数据包。任何值都会原样应用 (0 也是合法的 mark)；不传该参数即保持 SO_MARK 不变。需要 CAP_NET_ADMIN 权限。注意：FakeTCP 的实际载荷通过原始 TUN 写入，内核不会为其打标记；如有需要请在 TUN 设备上单独打标记。"
   enable_kcp_proxy:
     en: "proxy tcp streams with kcp, improving the latency and throughput on the network with udp packet loss."
     zh-CN: "使用 KCP 代理 TCP 流，提高在 UDP 丢包网络上的延迟和吞吐量。"

easytier/src/common/config.rs

@@ -73,6 +73,7 @@ pub fn gen_default_flags() -> Flags {
         disable_upnp: false,
         disable_relay_data: false,
         enable_udp_broadcast_relay: false,
+        socket_mark: None,
     }
 }
 
@@ -1259,6 +1260,57 @@ pub mod tests {
     use std::path::PathBuf;
     use tempfile::NamedTempFile;
 
+    #[test]
+    fn socket_mark_config_file_roundtrip_none_some_and_zero() {
+        // Omitting the flag leaves socket_mark unset (None) -> SO_MARK untouched.
+        let cfg = TomlConfigLoader::new_from_str(
+            r#"
+[network_identity]
+network_name = "n"
+network_secret = "s"
+"#,
+        )
+        .unwrap();
+        assert_eq!(cfg.get_flags().socket_mark, None);
+
+        // socket_mark = 0 is a legitimate value distinct from "unset".
+        let cfg = TomlConfigLoader::new_from_str(
+            r#"
+[network_identity]
+network_name = "n"
+network_secret = "s"
+
+[flags]
+socket_mark = 0
+"#,
+        )
+        .unwrap();
+        assert_eq!(cfg.get_flags().socket_mark, Some(0));
+
+        // A non-zero mark round-trips as Some(v).
+        let cfg = TomlConfigLoader::new_from_str(
+            r#"
+[network_identity]
+network_name = "n"
+network_secret = "s"
+
+[flags]
+socket_mark = 66
+"#,
+        )
+        .unwrap();
+        assert_eq!(cfg.get_flags().socket_mark, Some(66));
+
+        // set_flags(None) must serialize back through gen_config without
+        // resurrecting a value (guards the gen_flags merge against dropping
+        // the key when the serialized default is null).
+        cfg.set_flags(Flags {
+            socket_mark: None,
+            ..cfg.get_flags()
+        });
+        assert_eq!(cfg.get_flags().socket_mark, None);
+    }
+
     #[test]
     fn test_stun_servers_config() {
         let config = TomlConfigLoader::default();

easytier/src/connector/mod.rs

@@ -268,6 +268,7 @@ pub async fn create_connector_by_url(
                 IpScheme::FakeTcp => tunnel::fake_tcp::FakeTcpTunnelConnector::new(url).boxed(),
             };
             connector.set_resolved_addr(resolved_addr.addr);
+            connector.set_socket_mark(global_ctx.config.get_flags().socket_mark);
             if global_ctx.config.get_flags().bind_device {
                 set_bind_addr_for_peer_connector(
                     &mut connector,

easytier/src/core.rs

@@ -533,6 +533,17 @@ struct NetworkOptions {
     )]
     bind_device: Option<bool>,
 
+    // SO_MARK (fwmark) is a Linux-family kernel feature. Gate the flag out
+    // entirely on other targets so users on Windows/macOS/BSD don't see a
+    // `--socket-mark` they can't act on.
+    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+    #[arg(
+        long,
+        env = "ET_SOCKET_MARK",
+        help = t!("core_clap.socket_mark").to_string()
+    )]
+    socket_mark: Option<u32>,
+
     #[arg(
         long,
         env = "ET_ENABLE_KCP_PROXY",
@@ -1126,6 +1137,10 @@ impl NetworkOptions {
             .into();
         }
         f.bind_device = self.bind_device.unwrap_or(f.bind_device);
+        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+        {
+            f.socket_mark = self.socket_mark.or(f.socket_mark);
+        }
         f.enable_kcp_proxy = self.enable_kcp_proxy.unwrap_or(f.enable_kcp_proxy);
         f.disable_kcp_input = self.disable_kcp_input.unwrap_or(f.disable_kcp_input);
         f.enable_quic_proxy = self.enable_quic_proxy.unwrap_or(f.enable_quic_proxy);

easytier/src/instance/listeners.rs

@@ -27,10 +27,20 @@ pub fn create_listener_by_url(
     l: &url::Url,
     global_ctx: ArcGlobalCtx,
 ) -> Result<Box<dyn TunnelListener>, Error> {
+    use crate::common::config::ConfigLoader;
+    let socket_mark = global_ctx.config.get_flags().socket_mark;
     Ok(match l.try_into()? {
         TunnelScheme::Ip(scheme) => match scheme {
-            IpScheme::Tcp => TcpTunnelListener::new(l.clone()).boxed(),
-            IpScheme::Udp => UdpTunnelListener::new(l.clone()).boxed(),
+            IpScheme::Tcp => {
+                let mut l = TcpTunnelListener::new(l.clone());
+                l.set_socket_mark(socket_mark);
+                l.boxed()
+            }
+            IpScheme::Udp => {
+                let mut l = UdpTunnelListener::new(l.clone());
+                l.set_socket_mark(socket_mark);
+                l.boxed()
+            }
             #[cfg(feature = "wireguard")]
             IpScheme::Wg => {
                 use crate::tunnel::wireguard::{WgConfig, WgTunnelListener};
@@ -39,15 +49,20 @@ pub fn create_listener_by_url(
                     &nid.network_name,
                     &nid.network_secret.unwrap_or_default(),
                 );
-                WgTunnelListener::new(l.clone(), wg_config).boxed()
+                let mut l = WgTunnelListener::new(l.clone(), wg_config);
+                l.set_socket_mark(socket_mark);
+                l.boxed()
             }
             #[cfg(feature = "quic")]
             IpScheme::Quic => {
+                // QUIC reads socket_mark from global_ctx in QuicEndpointManager
                 tunnel::quic::QuicTunnelListener::new(l.clone(), global_ctx.clone()).boxed()
             }
             #[cfg(feature = "websocket")]
             IpScheme::Ws | IpScheme::Wss => {
-                tunnel::websocket::WsTunnelListener::new(l.clone()).boxed()
+                let mut l = tunnel::websocket::WsTunnelListener::new(l.clone());
+                l.set_socket_mark(socket_mark);
+                l.boxed()
             }
             #[cfg(feature = "faketcp")]
             IpScheme::FakeTcp => tunnel::fake_tcp::FakeTcpTunnelListener::new(l.clone()).boxed(),

easytier/src/launcher.rs

@@ -768,6 +768,10 @@ impl NetworkConfig {
             flags.bind_device = bind_device;
         }
 
+        if self.socket_mark.is_some() {
+            flags.socket_mark = self.socket_mark;
+        }
+
         if let Some(no_tun) = self.no_tun {
             flags.no_tun = no_tun;
         }
@@ -988,6 +992,7 @@ impl NetworkConfig {
         result.p2p_only = Some(flags.p2p_only);
         result.lazy_p2p = Some(flags.lazy_p2p);
         result.bind_device = Some(flags.bind_device);
+        result.socket_mark = flags.socket_mark;
         result.no_tun = Some(flags.no_tun);
         result.enable_exit_node = Some(flags.enable_exit_node);
         result.relay_all_peer_rpc = Some(flags.relay_all_peer_rpc);

easytier/src/peers/foreign_network_manager.rs

@@ -284,6 +284,10 @@ impl ForeignNetworkEntry {
         let mut flags = config.get_flags();
         flags.disable_relay_kcp = !global_ctx.get_flags().enable_relay_foreign_network_kcp;
         flags.disable_relay_quic = !global_ctx.get_flags().enable_relay_foreign_network_quic;
+        // socket_mark is a host-wide socket option: propagate from parent so
+        // outbound sockets the foreign-network entry initiates inherit the same
+        // mark as the rest of the node.
+        flags.socket_mark = global_ctx.get_flags().socket_mark;
         config.set_flags(flags);
 
         config.set_mapped_listeners(Some(global_ctx.config.get_mapped_listeners()));

easytier/src/proto/api_manage.proto

@@ -101,6 +101,7 @@ message NetworkConfig {
   optional string ipv6_public_addr_prefix = 64;
   optional bool disable_relay_data = 65;
   optional bool enable_udp_broadcast_relay = 66;
+  optional uint32 socket_mark = 67;
 }
 
 message PortForwardConfig {

easytier/src/proto/common.proto

@@ -77,6 +77,13 @@ message FlagsInConfig {
   bool disable_upnp = 40;
   bool disable_relay_data = 41;
   bool enable_udp_broadcast_relay = 42;
+
+  // Linux-only: SO_MARK (fwmark) value applied to every outbound underlay
+  // socket (TCP/UDP/QUIC/WS/WG connectors and listeners). Unset = leave
+  // SO_MARK untouched (kernel default 0). Any set value (including 0) is
+  // applied via setsockopt. Requires CAP_NET_ADMIN; silently ignored on
+  // non-Linux platforms.
+  optional uint32 socket_mark = 43;
 }
 
 message RpcDescriptor {

easytier/src/tunnel/common.rs

@@ -417,6 +417,7 @@ fn setup_socket2_ext(
     bind_addr: &SocketAddr,
     #[allow(unused_variables)] bind_dev: Option<String>,
     only_v6: bool,
+    socket_mark: Option<u32>,
 ) -> Result<(), TunnelError> {
     #[cfg(target_os = "windows")]
     {
@@ -430,6 +431,12 @@ fn setup_socket2_ext(
 
     socket2_socket.set_nonblocking(true)?;
     socket2_socket.set_reuse_address(!cfg!(target_os = "windows"))?;
+
+    // SO_MARK must be set before bind() so the kernel applies the mark to
+    // any source-address selection bind() triggers on unspecified binds.
+    // Accepted child sockets inherit the mark from the listener on Linux.
+    apply_socket_mark(socket2_socket, socket_mark)?;
+
     if let Err(e) = socket2_socket.bind(&socket2::SockAddr::from(*bind_addr)) {
         if bind_addr.is_ipv4() {
             return Err(e.into());
@@ -476,6 +483,29 @@ fn setup_socket2_ext(
     Ok(())
 }
 
+/// Apply Linux SO_MARK (a.k.a. fwmark) to a `socket2::Socket`. `None` leaves
+/// SO_MARK untouched (kernel default 0); `Some(mark)` applies that exact value
+/// — including `Some(0)`, which is a legitimate mark. On non-Linux platforms
+/// this is unconditionally a no-op.
+///
+/// Exposed so transports that bypass [`bind`] (currently the WebSocket
+/// default-bind path and FakeTCP) can apply the same mark.
+pub fn apply_socket_mark(
+    socket: &socket2::Socket,
+    socket_mark: Option<u32>,
+) -> Result<(), TunnelError> {
+    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+    if let Some(mark) = socket_mark {
+        tracing::trace!(socket_mark = mark, "set SO_MARK on socket");
+        socket.set_mark(mark)?;
+    }
+    #[cfg(not(any(target_os = "android", target_os = "fuchsia", target_os = "linux")))]
+    {
+        let _ = (socket, socket_mark);
+    }
+    Ok(())
+}
+
 #[derive(Debug, Default, Clone)]
 pub enum BindDev {
     #[default]
@@ -529,6 +559,9 @@ pub fn bind<B: Bindable>(
     #[builder(default, into)] dev: BindDev,
     net_ns: Option<NetNS>,
     #[builder(default)] only_v6: bool,
+    /// Linux SO_MARK (fwmark) to apply to the socket. `None` leaves SO_MARK
+    /// untouched; `Some(mark)` applies that exact value, including `Some(0)`.
+    socket_mark: Option<u32>,
 ) -> Result<B, TunnelError> {
     let _g = net_ns.map(|n| n.guard());
     let dev = match dev {
@@ -537,7 +570,7 @@ pub fn bind<B: Bindable>(
         BindDev::Custom(s) => Some(s),
     };
     let socket = socket2::Socket::new(socket2::Domain::for_address(addr), B::TYPE, B::PROTOCOL)?;
-    setup_socket2_ext(&socket, &addr, dev, only_v6)?;
+    setup_socket2_ext(&socket, &addr, dev, only_v6, socket_mark)?;
     B::finalize(socket)
 }
 
@@ -567,6 +600,59 @@ pub mod tests {
         packet_def::{PEER_MANAGER_HEADER_SIZE, TCP_TUNNEL_HEADER_SIZE},
     };
 
+    #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+    #[test]
+    fn apply_socket_mark_none_is_noop_and_does_not_error() {
+        // The contract for `None` is "no syscall is made and no error is
+        // returned" — must not require CAP_NET_ADMIN to call.
+        let socket = socket2::Socket::new(
+            socket2::Domain::IPV4,
+            socket2::Type::DGRAM,
+            Some(socket2::Protocol::UDP),
+        )
+        .unwrap();
+        super::apply_socket_mark(&socket, None).unwrap();
+    }
+
+    #[cfg(target_os = "linux")]
+    #[test]
+    #[ignore = "requires CAP_NET_ADMIN; run as root or with sudo -E cargo test"]
+    fn apply_socket_mark_sets_so_mark_when_capable() {
+        use nix::libc;
+        use std::os::fd::AsRawFd;
+
+        fn read_so_mark(s: &socket2::Socket) -> u32 {
+            let mut value: libc::c_int = 0;
+            let mut len = std::mem::size_of::<libc::c_int>() as libc::socklen_t;
+            let r = unsafe {
+                libc::getsockopt(
+                    s.as_raw_fd(),
+                    libc::SOL_SOCKET,
+                    libc::SO_MARK,
+                    &mut value as *mut _ as *mut libc::c_void,
+                    &mut len,
+                )
+            };
+            assert_eq!(r, 0, "getsockopt(SO_MARK) failed");
+            value as u32
+        }
+
+        let socket = socket2::Socket::new(
+            socket2::Domain::IPV4,
+            socket2::Type::DGRAM,
+            Some(socket2::Protocol::UDP),
+        )
+        .unwrap();
+        super::apply_socket_mark(&socket, Some(0x1234))
+            .expect("set_mark failed; need CAP_NET_ADMIN");
+        assert_eq!(read_so_mark(&socket), 0x1234);
+
+        // Some(0) is a legitimate value: it must reach setsockopt and clear
+        // the mark, distinct from None which makes no syscall.
+        super::apply_socket_mark(&socket, Some(0)).expect("set_mark(0) failed");
+        assert_eq!(read_so_mark(&socket), 0);
+    }
+
     #[test]
     fn framed_reader_rejects_short_peer_manager_body() {
         let mut buf = BytesMut::new();

easytier/src/tunnel/fake_tcp/mod.rs

@@ -282,6 +282,7 @@ pub struct FakeTcpTunnelConnector {
     addr: url::Url,
     ip_to_if_name: IpToIfNameCache,
     resolved_addr: Option<SocketAddr>,
+    socket_mark: Option<u32>,
 }
 
 impl FakeTcpTunnelConnector {
@@ -290,6 +291,7 @@ impl FakeTcpTunnelConnector {
             addr,
             ip_to_if_name: IpToIfNameCache::new(),
             resolved_addr: None,
+            socket_mark: None,
         }
     }
 }
@@ -324,6 +326,15 @@ impl TunnelConnector for FakeTcpTunnelConnector {
             .ok_or(TunnelError::InternalError("Failed to get local ip".into()))?;
 
         let os_socket = tokio::net::TcpSocket::new_v4()?;
+        // SO_MARK applies only to the kernel-visible "decoy" socket below.
+        // The actual FakeTCP payload travels via crafted segments written
+        // straight to the TUN device, which the kernel doesn't tag with
+        // SO_MARK. Operators relying on fwmark for FakeTCP must mark the
+        // TUN device's traffic with a separate nftables/iptables rule.
+        crate::tunnel::common::apply_socket_mark(
+            &socket2::SockRef::from(&os_socket),
+            self.socket_mark,
+        )?;
         os_socket.bind("0.0.0.0:0".parse().unwrap())?;
         let local_port = os_socket.local_addr()?.port();
         let local_addr = SocketAddr::new(local_ip, local_port);
@@ -399,6 +410,10 @@ impl TunnelConnector for FakeTcpTunnelConnector {
     fn set_resolved_addr(&mut self, addr: SocketAddr) {
         self.resolved_addr = Some(addr);
     }
+
+    fn set_socket_mark(&mut self, socket_mark: Option<u32>) {
+        self.socket_mark = socket_mark;
+    }
 }
 
 type RecvFut = Pin<Box<dyn Future<Output = Option<(BytesMut, usize)>> + Send + Sync>>;

easytier/src/tunnel/mod.rs

@@ -142,6 +142,10 @@ pub trait TunnelConnector: Send {
     fn set_bind_addrs(&mut self, _addrs: Vec<SocketAddr>) {}
     fn set_ip_version(&mut self, _ip_version: IpVersion) {}
     fn set_resolved_addr(&mut self, _addr: SocketAddr) {}
+    /// Linux SO_MARK to apply to outbound sockets. `None` leaves SO_MARK
+    /// untouched; `Some(mark)` applies that exact value (including `Some(0)`).
+    /// Default impl is a no-op; IP-based connectors override.
+    fn set_socket_mark(&mut self, _socket_mark: Option<u32>) {}
 }
 
 pub fn build_url_from_socket_addr(addr: &String, scheme: &str) -> url::Url {

easytier/src/tunnel/quic.rs

@@ -230,10 +230,15 @@ pub struct QuicEndpointManager {
 static QUIC_ENDPOINT_MANAGER: OnceLock<QuicEndpointManager> = OnceLock::new();
 
 impl QuicEndpointManager {
-    fn try_create(addr: SocketAddr, dual_stack: bool) -> Result<Endpoint, TunnelError> {
+    fn try_create(
+        addr: SocketAddr,
+        dual_stack: bool,
+        socket_mark: Option<u32>,
+    ) -> Result<Endpoint, TunnelError> {
         let socket = bind::<UdpSocket>()
             .addr(addr)
             .only_v6(addr.is_ipv6() && !dual_stack)
+            .maybe_socket_mark(socket_mark)
             .call()?;
         let runtime = default_runtime().ok_or(TunnelError::InternalError(
             "no async runtime found".to_owned(),
@@ -250,6 +255,7 @@ impl QuicEndpointManager {
 
     fn create<F>(
         &self,
+        socket_mark: Option<u32>,
         mut selector: F,
     ) -> Result<(&RwPool<Endpoint>, Option<Endpoint>), TunnelError>
     where
@@ -261,7 +267,7 @@ impl QuicEndpointManager {
                 return Ok((pool, None));
             };
 
-            let endpoint = Self::try_create(addr, dual_stack);
+            let endpoint = Self::try_create(addr, dual_stack, socket_mark);
             if let Err(error) = endpoint.as_ref()
                 && dual_stack
             {
@@ -331,8 +337,9 @@ impl QuicEndpointManager {
     /// * `addr`: listen address
     fn server(global_ctx: &ArcGlobalCtx, addr: SocketAddr) -> Result<Endpoint, TunnelError> {
         let mgr = Self::load(global_ctx);
+        let socket_mark = global_ctx.config.get_flags().socket_mark;
 
-        let (pool, endpoint) = mgr.create(|mgr| {
+        let (pool, endpoint) = mgr.create(socket_mark, |mgr| {
             let dual_stack = addr.ip() == Ipv6Addr::UNSPECIFIED && mgr.both.is_enabled();
             let pool = if addr.is_ipv4() {
                 &mgr.ipv4
@@ -351,8 +358,12 @@ impl QuicEndpointManager {
         Ok(endpoint)
     }
 
-    fn client_endpoint(&self, ip_version: IpVersion) -> Result<Endpoint, TunnelError> {
-        let (pool, endpoint) = self.create(|mgr| {
+    fn client_endpoint(
+        &self,
+        ip_version: IpVersion,
+        socket_mark: Option<u32>,
+    ) -> Result<Endpoint, TunnelError> {
+        let (pool, endpoint) = self.create(socket_mark, |mgr| {
             let dual_stack = mgr.both.is_enabled();
             let (pool, addr) = match ip_version {
                 IpVersion::V4 if !dual_stack => (&mgr.ipv4, (Ipv4Addr::UNSPECIFIED, 0).into()),
@@ -404,8 +415,9 @@ impl QuicEndpointManager {
         } else {
             IpVersion::V6
         };
+        let socket_mark = global_ctx.config.get_flags().socket_mark;
         Self::load(global_ctx)
-            .connect_with_ip_version(addr, ip_version)
+            .connect_with_ip_version(addr, ip_version, socket_mark)
             .await
     }
 
@@ -413,12 +425,13 @@ impl QuicEndpointManager {
         &self,
         addr: SocketAddr,
         ip_version: IpVersion,
+        socket_mark: Option<u32>,
     ) -> Result<(Endpoint, Connection), TunnelError> {
         let max_endpoint_stopping_retries = self.client_pool(ip_version).len().saturating_add(1);
         let mut endpoint_stopping_retries = 0;
 
         loop {
-            let endpoint = self.client_endpoint(ip_version)?;
+            let endpoint = self.client_endpoint(ip_version, socket_mark)?;
             let connecting = match endpoint.connect(addr, "localhost") {
                 Ok(connecting) => connecting,
                 Err(ConnectError::EndpointStopping) => {
@@ -646,7 +659,7 @@ mod tests {
     fn stopped_client_endpoint() -> (Endpoint, SocketAddr) {
         let rt = Builder::new_current_thread().enable_all().build().unwrap();
         let endpoint = rt.block_on(async {
-            QuicEndpointManager::try_create((Ipv4Addr::UNSPECIFIED, 0).into(), false).unwrap()
+            QuicEndpointManager::try_create((Ipv4Addr::UNSPECIFIED, 0).into(), false, None).unwrap()
         });
         let local_addr = endpoint.local_addr().unwrap();
         drop(rt);
@@ -763,7 +776,7 @@ mod tests {
             assert!(mgr.contains_local_addr(stopped_addr_b));
 
             let err = mgr
-                .connect_with_ip_version("127.0.0.1:0".parse().unwrap(), IpVersion::V4)
+                .connect_with_ip_version("127.0.0.1:0".parse().unwrap(), IpVersion::V4, None)
                 .await
                 .unwrap_err();
             let err = format!("{:?}", err);

easytier/src/tunnel/tcp.rs

@@ -1,7 +1,7 @@
 use std::net::SocketAddr;
 
 use super::{FromUrl, TunnelInfo};
-use crate::tunnel::common::bind;
+use crate::tunnel::common::{apply_socket_mark, bind};
 use async_trait::async_trait;
 use futures::stream::FuturesUnordered;
 use tokio::net::{TcpListener, TcpSocket, TcpStream};
@@ -17,6 +17,7 @@ const TCP_MTU_BYTES: usize = 2000;
 pub struct TcpTunnelListener {
     addr: url::Url,
     listener: Option<TcpListener>,
+    socket_mark: Option<u32>,
 }
 
 impl TcpTunnelListener {
@@ -24,9 +25,14 @@ impl TcpTunnelListener {
         TcpTunnelListener {
             addr,
             listener: None,
+            socket_mark: None,
         }
     }
 
+    pub fn set_socket_mark(&mut self, socket_mark: Option<u32>) {
+        self.socket_mark = socket_mark;
+    }
+
     async fn do_accept(&self) -> Result<Box<dyn Tunnel>, std::io::Error> {
         let listener = self.listener.as_ref().unwrap();
         let (stream, _) = listener.accept().await?;
@@ -61,7 +67,11 @@ impl TunnelListener for TcpTunnelListener {
         self.listener = None;
 
         let addr = SocketAddr::from_url(self.addr.clone(), IpVersion::Both).await?;
-        let listener = bind::<TcpListener>().addr(addr).only_v6(true).call()?;
+        let listener = bind::<TcpListener>()
+            .addr(addr)
+            .only_v6(true)
+            .maybe_socket_mark(self.socket_mark)
+            .call()?;
 
         self.addr
             .set_port(Some(listener.local_addr()?.port()))
@@ -130,6 +140,7 @@ pub struct TcpTunnelConnector {
     bind_addrs: Vec<SocketAddr>,
     ip_version: IpVersion,
     resolved_addr: Option<SocketAddr>,
+    socket_mark: Option<u32>,
 }
 
 impl TcpTunnelConnector {
@@ -139,6 +150,7 @@ impl TcpTunnelConnector {
             bind_addrs: vec![],
             ip_version: IpVersion::Both,
             resolved_addr: None,
+            socket_mark: None,
         }
     }
 
@@ -147,7 +159,19 @@ impl TcpTunnelConnector {
         addr: SocketAddr,
     ) -> Result<Box<dyn Tunnel>, super::TunnelError> {
         tracing::info!(url = ?self.addr, ?addr, "connect tcp start, bind addrs: {:?}", self.bind_addrs);
-        let stream = TcpStream::connect(addr).await?;
+        let stream = if self.socket_mark.is_some() {
+            // SO_MARK requires applying the option on the socket before
+            // connect, so go through TcpSocket rather than TcpStream::connect.
+            let socket = if addr.is_ipv4() {
+                TcpSocket::new_v4()?
+            } else {
+                TcpSocket::new_v6()?
+            };
+            apply_socket_mark(&socket2::SockRef::from(&socket), self.socket_mark)?;
+            socket.connect(addr).await?
+        } else {
+            TcpStream::connect(addr).await?
+        };
         tracing::info!(url = ?self.addr, ?addr, "connect tcp succ");
         get_tunnel_with_tcp_stream(stream, self.addr.clone())
     }
@@ -160,7 +184,12 @@ impl TcpTunnelConnector {
 
         for bind_addr in self.bind_addrs.iter() {
             tracing::info!(?bind_addr, ?addr, "bind addr");
-            match bind::<TcpSocket>().addr(*bind_addr).only_v6(true).call() {
+            match bind::<TcpSocket>()
+                .addr(*bind_addr)
+                .only_v6(true)
+                .maybe_socket_mark(self.socket_mark)
+                .call()
+            {
                 Ok(socket) => futures.push(socket.connect(addr)),
                 Err(error) => {
                     tracing::error!(?bind_addr, ?addr, ?error, "bind addr fail");
@@ -203,6 +232,10 @@ impl super::TunnelConnector for TcpTunnelConnector {
     fn set_resolved_addr(&mut self, addr: SocketAddr) {
         self.resolved_addr = Some(addr);
     }
+
+    fn set_socket_mark(&mut self, socket_mark: Option<u32>) {
+        self.socket_mark = socket_mark;
+    }
 }
 
 #[cfg(test)]

easytier/src/tunnel/udp.rs

@@ -567,6 +567,7 @@ pub struct UdpTunnelListener {
     data: UdpTunnelListenerData,
     forward_tasks: Arc<std::sync::Mutex<JoinSet<()>>>,
     close_event_recv: Option<UdpCloseEventReceiver>,
+    socket_mark: Option<u32>,
 }
 
 impl UdpTunnelListener {
@@ -580,9 +581,14 @@ impl UdpTunnelListener {
             data: UdpTunnelListenerData::new(addr, conn_send, close_event_send),
             forward_tasks: Arc::new(std::sync::Mutex::new(JoinSet::new())),
             close_event_recv: Some(close_event_recv),
+            socket_mark: None,
         }
     }
 
+    pub fn set_socket_mark(&mut self, socket_mark: Option<u32>) {
+        self.socket_mark = socket_mark;
+    }
+
     pub fn new_with_socket(addr: url::Url, socket: Arc<UdpSocket>) -> Self {
         let mut listener = Self::new(addr);
         listener.socket = Some(socket);
@@ -605,6 +611,7 @@ impl TunnelListener for UdpTunnelListener {
                     .addr(addr)
                     .only_v6(true)
                     .maybe_dev(tunnel_url.bind_dev())
+                    .maybe_socket_mark(self.socket_mark)
                     .call()?,
             ));
         }
@@ -683,6 +690,7 @@ pub struct UdpTunnelConnector {
     bind_addrs: Vec<SocketAddr>,
     ip_version: IpVersion,
     resolved_addr: Option<SocketAddr>,
+    socket_mark: Option<u32>,
 }
 
 impl UdpTunnelConnector {
@@ -692,6 +700,7 @@ impl UdpTunnelConnector {
             bind_addrs: vec![],
             ip_version: IpVersion::Both,
             resolved_addr: None,
+            socket_mark: None,
         }
     }
 
@@ -876,11 +885,18 @@ impl UdpTunnelConnector {
         &self,
         addr: SocketAddr,
     ) -> Result<Box<dyn Tunnel>, super::TunnelError> {
-        let socket = if addr.is_ipv4() {
-            UdpSocket::bind("0.0.0.0:0").await?
+        // Route through bind() so socket_mark is applied consistently for
+        // both the None (no-op) and Some(_) paths.
+        let bind_addr: SocketAddr = if addr.is_ipv4() {
+            "0.0.0.0:0".parse().unwrap()
         } else {
-            UdpSocket::bind("[::]:0").await?
+            "[::]:0".parse().unwrap()
         };
+        let socket = bind::<UdpSocket>()
+            .addr(bind_addr)
+            .only_v6(true)
+            .maybe_socket_mark(self.socket_mark)
+            .call()?;
 
         return self.try_connect_with_socket(Arc::new(socket), addr).await;
     }
@@ -893,7 +909,12 @@ impl UdpTunnelConnector {
 
         for bind_addr in self.bind_addrs.iter() {
             tracing::info!(?bind_addr, ?addr, "bind addr");
-            match bind().addr(*bind_addr).only_v6(true).call() {
+            match bind()
+                .addr(*bind_addr)
+                .only_v6(true)
+                .maybe_socket_mark(self.socket_mark)
+                .call()
+            {
                 Ok(socket) => futures.push(self.try_connect_with_socket(Arc::new(socket), addr)),
                 Err(error) => {
                     tracing::error!(?error, ?bind_addr, ?addr, "bind addr fail");
@@ -934,6 +955,10 @@ impl super::TunnelConnector for UdpTunnelConnector {
     fn set_resolved_addr(&mut self, addr: SocketAddr) {
         self.resolved_addr = Some(addr);
     }
+
+    fn set_socket_mark(&mut self, socket_mark: Option<u32>) {
+        self.socket_mark = socket_mark;
+    }
 }
 
 #[cfg(test)]

easytier/src/tunnel/websocket.rs

@@ -81,6 +81,7 @@ static TRUSTED_PROXIES: LazyLock<Vec<IpNetwork>> = LazyLock::new(|| {
 pub struct WsTunnelListener {
     addr: url::Url,
     listener: Option<TcpListener>,
+    socket_mark: Option<u32>,
 }
 
 impl WsTunnelListener {
@@ -88,9 +89,14 @@ impl WsTunnelListener {
         WsTunnelListener {
             addr,
             listener: None,
+            socket_mark: None,
         }
     }
 
+    pub fn set_socket_mark(&mut self, socket_mark: Option<u32>) {
+        self.socket_mark = socket_mark;
+    }
+
     async fn try_accept(&self, stream: TcpStream) -> Result<Box<dyn Tunnel>, TunnelError> {
         let peer_addr = stream.peer_addr()?;
         let mut remote_addr =
@@ -164,7 +170,11 @@ impl TunnelListener for WsTunnelListener {
         self.listener = None;
 
         let addr = SocketAddr::from_url(self.addr.clone(), IpVersion::Both).await?;
-        let listener = bind::<TcpListener>().addr(addr).only_v6(true).call()?;
+        let listener = bind::<TcpListener>()
+            .addr(addr)
+            .only_v6(true)
+            .maybe_socket_mark(self.socket_mark)
+            .call()?;
 
         self.addr
             .set_port(Some(listener.local_addr()?.port()))
@@ -201,6 +211,7 @@ pub struct WsTunnelConnector {
     resolved_addr: Option<SocketAddr>,
 
     bind_addrs: Vec<SocketAddr>,
+    socket_mark: Option<u32>,
 }
 
 impl WsTunnelConnector {
@@ -211,6 +222,7 @@ impl WsTunnelConnector {
             resolved_addr: None,
 
             bind_addrs: vec![],
+            socket_mark: None,
         }
     }
 
@@ -274,6 +286,10 @@ impl WsTunnelConnector {
         } else {
             TcpSocket::new_v6()?
         };
+        crate::tunnel::common::apply_socket_mark(
+            &socket2::SockRef::from(&socket),
+            self.socket_mark,
+        )?;
         Self::connect_with(self.addr.clone(), addr, socket).await
     }
 
@@ -285,7 +301,12 @@ impl WsTunnelConnector {
 
         for bind_addr in self.bind_addrs.iter() {
             tracing::info!(?bind_addr, ?addr, "bind addr");
-            match bind().addr(*bind_addr).only_v6(true).call() {
+            match bind()
+                .addr(*bind_addr)
+                .only_v6(true)
+                .maybe_socket_mark(self.socket_mark)
+                .call()
+            {
                 Ok(socket) => futures.push(Self::connect_with(self.addr.clone(), addr, socket)),
                 Err(error) => {
                     tracing::error!(?bind_addr, ?addr, ?error, "bind addr fail");
@@ -327,6 +348,10 @@ impl TunnelConnector for WsTunnelConnector {
     fn set_resolved_addr(&mut self, addr: SocketAddr) {
         self.resolved_addr = Some(addr);
     }
+
+    fn set_socket_mark(&mut self, socket_mark: Option<u32>) {
+        self.socket_mark = socket_mark;
+    }
 }
 
 #[cfg(test)]

easytier/src/tunnel/wireguard.rs

@@ -463,6 +463,7 @@ pub struct WgTunnelListener {
     wg_peer_map: Arc<DashMap<SocketAddr, Arc<WgPeer>>>,
 
     tasks: JoinSet<()>,
+    socket_mark: Option<u32>,
 }
 
 impl WgTunnelListener {
@@ -479,9 +480,14 @@ impl WgTunnelListener {
             wg_peer_map: Arc::new(DashMap::new()),
 
             tasks: JoinSet::new(),
+            socket_mark: None,
         }
     }
 
+    pub fn set_socket_mark(&mut self, socket_mark: Option<u32>) {
+        self.socket_mark = socket_mark;
+    }
+
     fn get_udp_socket(&self) -> Arc<UdpSocket> {
         self.udp.as_ref().unwrap().clone()
     }
@@ -561,6 +567,7 @@ impl TunnelListener for WgTunnelListener {
                 .addr(addr)
                 .only_v6(true)
                 .maybe_dev(tunnel_url.bind_dev())
+                .maybe_socket_mark(self.socket_mark)
                 .call()?,
         ));
         self.addr
@@ -599,6 +606,7 @@ pub struct WgTunnelConnector {
     bind_addrs: Vec<SocketAddr>,
     ip_version: IpVersion,
     resolved_addr: Option<SocketAddr>,
+    socket_mark: Option<u32>,
 }
 
 impl Debug for WgTunnelConnector {
@@ -619,6 +627,7 @@ impl WgTunnelConnector {
             bind_addrs: vec![],
             ip_version: IpVersion::Both,
             resolved_addr: None,
+            socket_mark: None,
         }
     }
 
@@ -695,6 +704,7 @@ impl WgTunnelConnector {
             .addr("[::]:0".parse().unwrap())
             .dev(BindDev::Disabled)
             .only_v6(true)
+            .maybe_socket_mark(self.socket_mark)
             .call()?;
         Self::connect_with_socket(self.addr.clone(), self.config.clone(), socket, addr).await
     }
@@ -721,7 +731,12 @@ impl super::TunnelConnector for WgTunnelConnector {
         let futures = FuturesUnordered::new();
         for bind_addr in bind_addrs.into_iter() {
             tracing::info!(?bind_addr, ?addr, "bind addr");
-            match bind().addr(bind_addr).only_v6(true).call() {
+            match bind()
+                .addr(bind_addr)
+                .only_v6(true)
+                .maybe_socket_mark(self.socket_mark)
+                .call()
+            {
                 Ok(socket) => futures.push(Self::connect_with_socket(
                     self.addr.clone(),
                     self.config.clone(),
@@ -753,6 +768,10 @@ impl super::TunnelConnector for WgTunnelConnector {
     fn set_resolved_addr(&mut self, addr: SocketAddr) {
         self.resolved_addr = Some(addr);
     }
+
+    fn set_socket_mark(&mut self, socket_mark: Option<u32>) {
+        self.socket_mark = socket_mark;
+    }
 }
 
 #[cfg(test)]