trojan udp seemingly working

.cargo/config

@@ -4,4 +4,4 @@ rustflags = ["--cfg", "tokio_unstable"]
 
 [env]
 BORING_BSSL_SOURCE_PATH = { value = "deps/boringssl/src", relative = true}
-RUST_LOG= { value = "clash=trace" }
+RUST_LOG= { value = "clash=debug" }

clash_lib/src/app/dispatcher/dispatcher.rs

@@ -342,7 +342,7 @@ impl Dispatcher {
 
         tokio::spawn(async move {
             let _ = close_receiver.await;
-            debug!("UDP close signal for {} received", s);
+            trace!("UDP close signal for {} received", s);
             t1.abort();
             t2.abort();
         });

clash_lib/src/proxy/socks/inbound/stream.rs

@@ -14,7 +14,7 @@ use std::{io, str};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
 use tokio_util::udp::UdpFramed;
-use tracing::{debug, info, warn};
+use tracing::{trace, warn};
 
 pub async fn handle_tcp<'a>(
     sess: &'a mut Session,
@@ -120,7 +120,7 @@ pub async fn handle_tcp<'a>(
 
     match buf[1] {
         socks_command::CONNECT => {
-            debug!("Got a CONNECT request from {}", s.peer_addr()?);
+            trace!("Got a CONNECT request from {}", s.peer_addr()?);
 
             buf.clear();
             buf.put_u8(SOCKS5_VERSION);
@@ -145,7 +145,7 @@ pub async fn handle_tcp<'a>(
             )
             .await?;
 
-            debug!(
+            trace!(
                 "Got a UDP_ASSOCIATE request from {}, UDP assigned at {}",
                 s.peer_addr()?,
                 udp_inbound.local_addr()?
@@ -184,7 +184,7 @@ pub async fn handle_tcp<'a>(
             buf.resize(1, 0);
             match s.read(&mut buf[..]).await {
                 Ok(_) => {
-                    info!("UDP association finished, closing");
+                    trace!("UDP association finished, closing");
                 }
                 Err(e) => {
                     warn!("SOCKS client closed connection: {}", e);

clash_lib/src/proxy/trojan/datagram.rs

@@ -5,9 +5,9 @@ use std::{
     task::Poll,
 };
 
-use bytes::{BufMut, BytesMut};
+use bytes::{Buf, BufMut, BytesMut};
 use futures::{pin_mut, ready, Future, Sink, Stream};
-use tracing::{debug, instrument};
+use tracing::{debug, trace};
 
 use tokio::io::{AsyncReadExt, AsyncWrite};
 
@@ -94,21 +94,22 @@ impl Sink<UdpPacket> for OutboundDatagramTrojan {
             payload.put_slice(b"\r\n");
             payload.put_slice(&data);
 
-            let n = ready!(inner.as_mut().poll_write(cx, payload.as_ref()))?;
+            while payload.len() != 0 {
+                let n = ready!(inner.as_mut().poll_write(cx, payload.as_ref()))?;
+
+                payload.advance(n);
+
+                trace!(
+                    "written {} bytes to trojan stream, remaining {}, data len {}",
+                    n,
+                    payload.len(),
+                    data.len()
+                );
+            }
 
-            let wrote_all = n == data.len();
-            *pkt_container = None;
             *flushed = true;
 
-            let res = if wrote_all {
-                Ok(())
-            } else {
-                Err(io::Error::new(
-                    io::ErrorKind::Other,
-                    "failed to write entire datagram",
-                ))
-            };
-            Poll::Ready(res)
+            Poll::Ready(Ok(()))
         } else {
             debug!("no udp packet to send");
             Poll::Ready(Err(io::Error::new(
@@ -314,11 +315,18 @@ impl Stream for OutboundDatagramTrojan {
                                 debug!("invalid trojan data");
                                 return Poll::Ready(None);
                             }
-                            let data = read_buf.split_to(*len);
+
+                            let addr = addr.to_owned();
+                            let len = len.to_owned();
+
+                            *state = ReadState::Atyp;
+
+                            let data = read_buf.split_to(len);
+
                             return Poll::Ready(Some(UdpPacket {
                                 data: data.to_vec(),
                                 src_addr: remote_addr.clone(),
-                                dst_addr: addr.to_owned(),
+                                dst_addr: addr,
                             }));
                         }
                         Err(err) => {

clash_lib/src/proxy/trojan/mod.rs

@@ -58,6 +58,41 @@ impl Handler {
     pub fn new(opts: Opts) -> AnyOutboundHandler {
         Arc::new(Self { opts })
     }
+
+    /// TCP: 0x01,
+    /// UDP: 0x03,
+    async fn inner_proxy_stream(
+        &self,
+        s: AnyStream,
+        sess: &Session,
+        tcp: bool,
+    ) -> io::Result<AnyStream> {
+        let tls_opt = TLSOptions {
+            skip_cert_verify: self.opts.skip_cert_verify,
+            sni: self.opts.sni.clone(),
+            alpn: self.opts.alpn.clone().or(Some(
+                DEFAULT_ALPN
+                    .to_vec()
+                    .into_iter()
+                    .map(|x| x.to_owned())
+                    .collect::<Vec<String>>(),
+            )),
+        };
+
+        let mut s = transport::tls::wrap_stream(s, tls_opt.to_owned()).await?;
+
+        let mut buf = BytesMut::new();
+        let password = Sha224::digest(self.opts.password.as_bytes());
+        let password = utils::encode_hex(&password[..]);
+        buf.put_slice(password.as_bytes());
+        buf.put_slice(b"\r\n");
+        buf.put_u8(if tcp { 0x01 } else { 0x03 }); // tcp
+        sess.destination.write_buf(&mut buf);
+        buf.put_slice(b"\r\n");
+        s.write_all(&buf).await?;
+
+        Ok(s)
+    }
 }
 
 #[async_trait]
@@ -115,31 +150,7 @@ impl OutboundHandler for Handler {
         sess: &Session,
         _: ThreadSafeDNSResolver,
     ) -> io::Result<AnyStream> {
-        let tls_opt = TLSOptions {
-            skip_cert_verify: self.opts.skip_cert_verify,
-            sni: self.opts.sni.clone(),
-            alpn: self.opts.alpn.clone().or(Some(
-                DEFAULT_ALPN
-                    .to_vec()
-                    .into_iter()
-                    .map(|x| x.to_owned())
-                    .collect::<Vec<String>>(),
-            )),
-        };
-
-        let mut s = transport::tls::wrap_stream(s, tls_opt.to_owned()).await?;
-
-        let mut buf = BytesMut::new();
-        let password = Sha224::digest(self.opts.password.as_bytes());
-        let password = utils::encode_hex(&password[..]);
-        buf.put_slice(password.as_bytes());
-        buf.put_slice(b"\r\n");
-        buf.put_u8(0x01); // tcp
-        sess.destination.write_buf(&mut buf);
-        buf.put_slice(b"\r\n");
-        s.write_all(&buf).await?;
-
-        Ok(s)
+        self.inner_proxy_stream(s, sess, true).await
     }
 
     async fn connect_datagram(
@@ -166,7 +177,8 @@ impl OutboundHandler for Handler {
         })
         .await?;
 
-        let stream = self.proxy_stream(stream, sess, resolver).await?;
+        let stream = self.inner_proxy_stream(stream, sess, false).await?;
+
         Ok(Box::new(OutboundDatagramTrojan::new(
             stream,
             sess.destination.clone(),