From e3ca7ffa543fb8f61eaf9237da124143b5a7d48b Mon Sep 17 00:00:00 2001 From: X <50459973+ly4096x@users.noreply.github.com> Date: Tue, 2 Jun 2026 21:57:18 -0400 Subject: [PATCH] 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 --- easytier/locales/app.yml | 3 + easytier/src/common/config.rs | 52 +++++++++++ easytier/src/connector/mod.rs | 1 + easytier/src/core.rs | 15 ++++ easytier/src/instance/listeners.rs | 23 ++++- easytier/src/launcher.rs | 5 ++ easytier/src/peers/foreign_network_manager.rs | 4 + easytier/src/proto/api_manage.proto | 1 + easytier/src/proto/common.proto | 7 ++ easytier/src/tunnel/common.rs | 88 ++++++++++++++++++- easytier/src/tunnel/fake_tcp/mod.rs | 15 ++++ easytier/src/tunnel/mod.rs | 4 + easytier/src/tunnel/quic.rs | 31 +++++-- easytier/src/tunnel/tcp.rs | 41 ++++++++- easytier/src/tunnel/udp.rs | 33 ++++++- easytier/src/tunnel/websocket.rs | 29 +++++- easytier/src/tunnel/wireguard.rs | 21 ++++- 17 files changed, 348 insertions(+), 25 deletions(-) diff --git a/easytier/locales/app.yml b/easytier/locales/app.yml index 710bb4a7..cb4256ee 100644 --- a/easytier/locales/app.yml +++ b/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 丢包网络上的延迟和吞吐量。" diff --git a/easytier/src/common/config.rs b/easytier/src/common/config.rs index f92a3100..8eb9f957 100644 --- a/easytier/src/common/config.rs +++ b/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(); diff --git a/easytier/src/connector/mod.rs b/easytier/src/connector/mod.rs index bfc6c7f2..a5e72b89 100644 --- a/easytier/src/connector/mod.rs +++ b/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, diff --git a/easytier/src/core.rs b/easytier/src/core.rs index 85875135..afde3dea 100644 --- a/easytier/src/core.rs +++ b/easytier/src/core.rs @@ -533,6 +533,17 @@ struct NetworkOptions { )] bind_device: Option, + // 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, + #[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); diff --git a/easytier/src/instance/listeners.rs b/easytier/src/instance/listeners.rs index 72a1b9af..c0ecc60c 100644 --- a/easytier/src/instance/listeners.rs +++ b/easytier/src/instance/listeners.rs @@ -27,10 +27,20 @@ pub fn create_listener_by_url( l: &url::Url, global_ctx: ArcGlobalCtx, ) -> Result, 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(), diff --git a/easytier/src/launcher.rs b/easytier/src/launcher.rs index 12815d84..f4f4649e 100644 --- a/easytier/src/launcher.rs +++ b/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); diff --git a/easytier/src/peers/foreign_network_manager.rs b/easytier/src/peers/foreign_network_manager.rs index 517137a3..c34240cc 100644 --- a/easytier/src/peers/foreign_network_manager.rs +++ b/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())); diff --git a/easytier/src/proto/api_manage.proto b/easytier/src/proto/api_manage.proto index 418cd7aa..978042ca 100644 --- a/easytier/src/proto/api_manage.proto +++ b/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 { diff --git a/easytier/src/proto/common.proto b/easytier/src/proto/common.proto index 308796e8..7a0bf404 100644 --- a/easytier/src/proto/common.proto +++ b/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 { diff --git a/easytier/src/tunnel/common.rs b/easytier/src/tunnel/common.rs index 874de258..03a75acf 100644 --- a/easytier/src/tunnel/common.rs +++ b/easytier/src/tunnel/common.rs @@ -417,6 +417,7 @@ fn setup_socket2_ext( bind_addr: &SocketAddr, #[allow(unused_variables)] bind_dev: Option, only_v6: bool, + socket_mark: Option, ) -> 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, +) -> 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( #[builder(default, into)] dev: BindDev, net_ns: Option, #[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, ) -> Result { let _g = net_ns.map(|n| n.guard()); let dev = match dev { @@ -537,7 +570,7 @@ pub fn bind( 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::() 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(); diff --git a/easytier/src/tunnel/fake_tcp/mod.rs b/easytier/src/tunnel/fake_tcp/mod.rs index 8acfd5a0..0f27d84f 100644 --- a/easytier/src/tunnel/fake_tcp/mod.rs +++ b/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, + socket_mark: Option, } 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) { + self.socket_mark = socket_mark; + } } type RecvFut = Pin> + Send + Sync>>; diff --git a/easytier/src/tunnel/mod.rs b/easytier/src/tunnel/mod.rs index 06629b6c..2f0e8032 100644 --- a/easytier/src/tunnel/mod.rs +++ b/easytier/src/tunnel/mod.rs @@ -142,6 +142,10 @@ pub trait TunnelConnector: Send { fn set_bind_addrs(&mut self, _addrs: Vec) {} 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) {} } pub fn build_url_from_socket_addr(addr: &String, scheme: &str) -> url::Url { diff --git a/easytier/src/tunnel/quic.rs b/easytier/src/tunnel/quic.rs index c4304dae..a9ad21e0 100644 --- a/easytier/src/tunnel/quic.rs +++ b/easytier/src/tunnel/quic.rs @@ -230,10 +230,15 @@ pub struct QuicEndpointManager { static QUIC_ENDPOINT_MANAGER: OnceLock = OnceLock::new(); impl QuicEndpointManager { - fn try_create(addr: SocketAddr, dual_stack: bool) -> Result { + fn try_create( + addr: SocketAddr, + dual_stack: bool, + socket_mark: Option, + ) -> Result { let socket = bind::() .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( &self, + socket_mark: Option, mut selector: F, ) -> Result<(&RwPool, Option), 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 { 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 { - let (pool, endpoint) = self.create(|mgr| { + fn client_endpoint( + &self, + ip_version: IpVersion, + socket_mark: Option, + ) -> Result { + 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, ) -> 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); diff --git a/easytier/src/tunnel/tcp.rs b/easytier/src/tunnel/tcp.rs index ec7de9be..8e678170 100644 --- a/easytier/src/tunnel/tcp.rs +++ b/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, + socket_mark: Option, } 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) { + self.socket_mark = socket_mark; + } + async fn do_accept(&self) -> Result, 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::().addr(addr).only_v6(true).call()?; + let listener = bind::() + .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, ip_version: IpVersion, resolved_addr: Option, + socket_mark: Option, } 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, 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::().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(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) { + self.socket_mark = socket_mark; + } } #[cfg(test)] diff --git a/easytier/src/tunnel/udp.rs b/easytier/src/tunnel/udp.rs index 714dd10f..b7ebd979 100644 --- a/easytier/src/tunnel/udp.rs +++ b/easytier/src/tunnel/udp.rs @@ -567,6 +567,7 @@ pub struct UdpTunnelListener { data: UdpTunnelListenerData, forward_tasks: Arc>>, close_event_recv: Option, + socket_mark: Option, } 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) { + self.socket_mark = socket_mark; + } + pub fn new_with_socket(addr: url::Url, socket: Arc) -> 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, ip_version: IpVersion, resolved_addr: Option, + socket_mark: Option, } 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, 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::() + .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) { + self.socket_mark = socket_mark; + } } #[cfg(test)] diff --git a/easytier/src/tunnel/websocket.rs b/easytier/src/tunnel/websocket.rs index e58736e0..1992fcfb 100644 --- a/easytier/src/tunnel/websocket.rs +++ b/easytier/src/tunnel/websocket.rs @@ -81,6 +81,7 @@ static TRUSTED_PROXIES: LazyLock> = LazyLock::new(|| { pub struct WsTunnelListener { addr: url::Url, listener: Option, + socket_mark: Option, } 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) { + self.socket_mark = socket_mark; + } + async fn try_accept(&self, stream: TcpStream) -> Result, 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::().addr(addr).only_v6(true).call()?; + let listener = bind::() + .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, bind_addrs: Vec, + socket_mark: Option, } 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) { + self.socket_mark = socket_mark; + } } #[cfg(test)] diff --git a/easytier/src/tunnel/wireguard.rs b/easytier/src/tunnel/wireguard.rs index c37d8d1f..e0711f2e 100644 --- a/easytier/src/tunnel/wireguard.rs +++ b/easytier/src/tunnel/wireguard.rs @@ -463,6 +463,7 @@ pub struct WgTunnelListener { wg_peer_map: Arc>>, tasks: JoinSet<()>, + socket_mark: Option, } 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) { + self.socket_mark = socket_mark; + } + fn get_udp_socket(&self) -> Arc { 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, ip_version: IpVersion, resolved_addr: Option, + socket_mark: Option, } 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) { + self.socket_mark = socket_mark; + } } #[cfg(test)]