feat(tproxy): support tproxy on Linux (#615)

clash/tests/data/config/rules.yaml

@@ -2,6 +2,7 @@
 port: 8888
 socks-port: 8889
 mixed-port: 8899
+tproxy-port: 8900
 
 tun:
   enable: false

clash/tests/data/config/tproxy.yaml

@@ -1,17 +1,9 @@
 ---
-port: 8888
-socks-port: 8889
-mixed-port: 8899
 tproxy-port: 8900
 
 mode: rule
 log-level: debug
 
-
-proxies:
-  - name: "tor"
-    type: tor
-
 rules:
-  - MATCH, tor
+  - MATCH, DIRECT
 ...

clash_lib/Cargo.toml

@@ -144,6 +144,9 @@ env_logger = "0.11"
 [build-dependencies]
 prost-build = "0.13"
 
+[target.'cfg(target_os="linux")'.dependencies]
+unix-udp-sock = { git = "https://github.com/Watfaq/unix-udp-sock.git", rev = "cd3e4eca43e6f3be82a2703c3d711b7e18fbfd18"}
+
 [target.'cfg(macos)'.dependencies]
 security-framework = "3.0.0"
 

clash_lib/src/app/inbound/manager.rs

@@ -6,7 +6,7 @@ use crate::{
         dispatcher::Dispatcher,
         inbound::network_listener::{ListenerType, NetworkInboundListener},
     },
-    common::auth::ThreadSafeAuthenticator,
+    common::{auth::ThreadSafeAuthenticator, errors::new_io_error},
     config::internal::config::{BindAddress, Inbound},
     Error, Runner,
 };
@@ -68,7 +68,21 @@ impl InboundManager {
         }
 
         Ok(Box::pin(async move {
-            futures::future::select_all(runners).await.0
+            let mut errors = Vec::new();
+            let _ = futures::future::join_all(runners)
+                .await
+                .into_iter()
+                .filter_map(|r| r.map_err(|e| errors.push(e)).ok())
+                .collect::<Vec<_>>();
+            if errors.is_empty() {
+                Ok(())
+            } else {
+                Err(new_io_error(format!(
+                    "failed to start inbound listeners: {:?}",
+                    errors
+                ))
+                .into())
+            }
         }))
     }
 

clash_lib/src/app/inbound/network_listener.rs

@@ -2,7 +2,10 @@ use crate::{
     common::auth::ThreadSafeAuthenticator, config::internal::config::BindAddress,
 };
 
-use crate::proxy::{http, mixed, socks, tproxy, AnyInboundListener};
+use crate::proxy::{http, mixed, socks, AnyInboundListener};
+
+#[cfg(target_os = "linux")]
+use crate::proxy::tproxy;
 
 use crate::{proxy::utils::Interface, Dispatcher, Error, Runner};
 use futures::FutureExt;
@@ -124,7 +127,18 @@ impl NetworkInboundListener {
                 self.authenticator.clone(),
             )),
             ListenerType::Tproxy => {
-                Arc::new(tproxy::Listener::new((ip, self.port).into()))
+                #[cfg(target_os = "linux")]
+                {
+                    Arc::new(tproxy::Listener::new(
+                        (ip, self.port).into(),
+                        self.dispatcher.clone(),
+                    ))
+                }
+                #[cfg(not(target_os = "linux"))]
+                {
+                    warn!("tproxy is not supported on this platform");
+                    return;
+                }
             }
         };
 

clash_lib/src/proxy/datagram.rs

@@ -1,20 +1,15 @@
 use crate::{
-    app::dns::ThreadSafeDNSResolver,
-    common::errors::new_io_error,
-    proxy::{socks::Socks5UDPCodec, InboundDatagram},
+    app::dns::ThreadSafeDNSResolver, common::errors::new_io_error,
     session::SocksAddr,
 };
-use bytes::Bytes;
-use futures::{ready, Sink, SinkExt, Stream, StreamExt};
+use futures::{ready, Sink, Stream};
 use std::{
     fmt::{Debug, Display, Formatter},
     io,
-    net::SocketAddr,
     pin::Pin,
     task::{Context, Poll},
 };
 use tokio::{io::ReadBuf, net::UdpSocket};
-use tokio_util::udp::UdpFramed;
 
 #[derive(Clone)]
 pub struct UdpPacket {
@@ -64,90 +59,6 @@ impl UdpPacket {
     }
 }
 
-pub struct InboundUdp<I> {
-    inner: I,
-}
-
-impl<I> InboundUdp<I>
-where
-    I: Stream + Unpin,
-    I: Sink<((Bytes, SocksAddr), SocketAddr)>,
-{
-    pub fn new(inner: I) -> Self {
-        Self { inner }
-    }
-}
-
-impl Debug for InboundUdp<UdpFramed<Socks5UDPCodec>> {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("InboundUdp").finish()
-    }
-}
-
-impl Stream for InboundUdp<UdpFramed<Socks5UDPCodec>> {
-    type Item = UdpPacket;
-
-    fn poll_next(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Option<Self::Item>> {
-        let pin = self.get_mut();
-
-        match pin.inner.poll_next_unpin(cx) {
-            Poll::Ready(item) => match item {
-                None => Poll::Ready(None),
-                Some(item) => match item {
-                    Ok(((dst, pkt), src)) => Poll::Ready(Some(UdpPacket {
-                        data: pkt.to_vec(),
-                        src_addr: SocksAddr::Ip(src),
-                        dst_addr: dst,
-                    })),
-                    Err(_) => Poll::Ready(None),
-                },
-            },
-            Poll::Pending => Poll::Pending,
-        }
-    }
-}
-
-impl Sink<UdpPacket> for InboundUdp<UdpFramed<Socks5UDPCodec>> {
-    type Error = std::io::Error;
-
-    fn poll_ready(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<(), Self::Error>> {
-        let pin = self.get_mut();
-        pin.inner.poll_ready_unpin(cx)
-    }
-
-    fn start_send(self: Pin<&mut Self>, item: UdpPacket) -> Result<(), Self::Error> {
-        let pin = self.get_mut();
-        pin.inner.start_send_unpin((
-            (item.data.into(), item.src_addr),
-            item.dst_addr.must_into_socket_addr(),
-        ))
-    }
-
-    fn poll_flush(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<(), Self::Error>> {
-        let pin = self.get_mut();
-        pin.inner.poll_flush_unpin(cx)
-    }
-
-    fn poll_close(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<(), Self::Error>> {
-        let pin = self.get_mut();
-        pin.inner.poll_close_unpin(cx)
-    }
-}
-
-impl InboundDatagram<UdpPacket> for InboundUdp<UdpFramed<Socks5UDPCodec>> {}
-
 #[must_use = "sinks do nothing unless polled"]
 // TODO: maybe we should use abstract datagram IO interface instead of the
 // Stream + Sink trait

clash_lib/src/proxy/mod.rs

@@ -28,6 +28,7 @@ pub mod reject;
 
 pub mod http;
 pub mod mixed;
+#[cfg(target_os = "linux")]
 pub mod tproxy;
 
 pub(crate) mod datagram;
@@ -82,6 +83,10 @@ pub trait InboundDatagram<Item>:
     Stream<Item = Item> + Sink<Item, Error = io::Error> + Send + Sync + Unpin + Debug
 {
 }
+impl<T, U> InboundDatagram<U> for T where
+    T: Stream<Item = U> + Sink<U, Error = io::Error> + Send + Sync + Unpin + Debug
+{
+}
 pub type AnyInboundDatagram =
     Box<dyn InboundDatagram<UdpPacket, Error = io::Error, Item = UdpPacket>>;