feat: relay refactor - support UDP relay (UoT only) (#374)

* wip

* add proxy connector

* fmt

* wip

* wip

* wip

* wip

* apply patch from #213

* clippy

* t

* t

clash/tests/data/config/rules.yaml

@@ -65,11 +65,32 @@ proxy-groups:
     proxies:
       - "plain-vmess"
       - "ws-vmess"
+      - "ss-simple"
+      - "trojan"
       - "auto"
       - "fallback-auto"
       - "load-balance"
       - "select"
+      - "wg"
       - DIRECT
+
+  - name: "udp-relay"
+    type: relay
+    proxies:
+      # - "plain-vmess"
+      - "ws-vmess"
+      # - "h2-vmess"
+      # - "tls-vmess"
+      # - "grpc-vmess"
+      # - "ss-simple"
+      # - "trojan"
+      # - "auto"
+      # - "fallback-auto"
+      # - "load-balance"
+      # - "select"
+      # - "wg"
+      # - DIRECT
+
 
   - name: "relay-one"
     type: relay
@@ -79,7 +100,7 @@ proxy-groups:
   - name: "auto"
     type: url-test
     use:
-      - "file-provider"
+      - "file-provider-uot"
     proxies:
       - DIRECT
     url: "http://www.gstatic.com/generate_204"
@@ -88,7 +109,7 @@ proxy-groups:
   - name: "fallback-auto"
     type: fallback
     use:
-      - "file-provider"
+      - "file-provider-uot"
     proxies:
       - DIRECT
     url: "http://www.gstatic.com/generate_204"
@@ -97,7 +118,7 @@ proxy-groups:
   - name: "load-balance"
     type: load-balance
     use:
-      - "file-provider"
+      - "file-provider-uot"
     proxies:
       - DIRECT
     strategy: round-robin
@@ -107,7 +128,7 @@ proxy-groups:
   - name: select
     type: select
     use:
-      - "file-provider"
+      - "file-provider-uot"
 
   - name: test 🌏
     type: select
@@ -215,6 +236,20 @@ proxies:
     grpc-opts:
       grpc-service-name: def
 
+  - name: "wg"
+    type: wireguard
+    server: engage.cloudflareclient.com
+    port: 2408
+    private-key: uIwDn4c7656E/1pHkJu23ZOe/4SuCnL+vL+jE2s4MHE=
+    ip: 172.16.0.2/32
+    ipv6: 2606:4700:110:8e5e:fa1:3f30:c077:e17c/128
+    public-key: bmXOC+F1FxEMF9dyiK2H5/1SUtzH0JuVo51h2wPfgyo=
+    allowed-ips: ['0.0.0.0/0', '::/0']
+    remote-dns-resolve: true
+    dns:
+      - 1.1.1.1
+    udp: true
+
 proxy-providers:
   file-provider:
     type: file
@@ -225,6 +260,15 @@ proxy-providers:
       url: http://www.gstatic.com/generate_204
       interval: 300
 
+  file-provider-uot:
+    type: file
+    path: ./uot.yaml
+    interval: 300
+    health-check:
+      enable: true
+      url: http://www.gstatic.com/generate_204
+      interval: 300
+
 rule-providers:
   file-provider:
     type: file
@@ -233,7 +277,7 @@ rule-providers:
     behavior: domain
 
 rules:
-  - DOMAIN,google.com,relay
+  - DOMAIN,google.com,ws-vmess
   - DOMAIN-KEYWORD,httpbin,trojan-grpc
   - DOMAIN,ipinfo.io,trojan-grpc
 #  - RULE-SET,file-provider,trojan
@@ -244,7 +288,7 @@ rules:
   - SRC-IP-CIDR,192.168.1.1/24,DIRECT
   - GEOIP,CN,DIRECT
   - IP-CIDR,10.0.0.11/32,select
-  - DST-PORT,53,trojan
+  - DST-PORT,53,ws-vmess
   - SRC-PORT,7777,DIRECT
   - MATCH, DIRECT
 ...

clash/tests/data/config/uot.yaml

@@ -0,0 +1,37 @@
+proxies:
+  - name: plain-vmess
+    type: vmess
+    server: 10.0.0.13
+    port: 16823
+    uuid: b831381d-6324-4d53-ad4f-8cda48b30811
+    alterId: 0
+    cipher: auto
+    udp: true
+    skip-cert-verify: true
+
+  - name: ws-vmess
+    type: vmess
+    server: 10.0.0.13
+    port: 16824
+    uuid: b831381d-6324-4d53-ad4f-8cda48b30811
+    alterId: 0
+    cipher: auto
+    udp: true
+    skip-cert-verify: true
+    network: ws
+    ws-opts:
+      path: /api/v3/download.getFile
+      headers:
+        Host: www.amazon.com
+
+  - name: "trojan"
+    type: trojan
+    server: 10.0.0.13
+    port: 9443
+    password: password1
+    udp: true
+    # sni: example.com # aka server name
+    alpn:
+      - h2
+      - http/1.1
+    skip-cert-verify: true

clash_lib/src/app/dispatcher/dispatcher_impl.rs

@@ -74,7 +74,7 @@ impl Dispatcher {
         *self.mode.lock().unwrap()
     }
 
-    #[instrument(skip(lhs))]
+    #[instrument(skip(self, sess, lhs))]
     pub async fn dispatch_stream<S>(&self, sess: Session, mut lhs: S)
     where
         S: AsyncRead + AsyncWrite + Unpin + Send,
@@ -124,11 +124,7 @@ impl Dispatcher {
 
         match handler
             .connect_stream(&sess, self.resolver.clone())
-            .instrument(info_span!(
-                "connect_stream",
-                outbound_name = outbound_name,
-                session = %sess,
-            ))
+            .instrument(info_span!("connect_stream", outbound_name = outbound_name,))
             .await
         {
             Ok(rhs) => {
@@ -145,7 +141,6 @@ impl Dispatcher {
                 .instrument(info_span!(
                     "copy_bidirectional",
                     outbound_name = outbound_name,
-                    session = %sess,
                 ))
                 .await
                 {

clash_lib/src/app/outbound/manager.rs

@@ -240,7 +240,7 @@ impl OutboundManager {
         ) -> Result<ThreadSafeProxyProvider, Error> {
             if name == PROXY_DIRECT || name == PROXY_REJECT {
                 return Err(Error::InvalidConfig(format!(
-                    "proxy group {} is reserved",
+                    "proxy group name `{}` is reserved",
                     name
                 )));
             }

clash_lib/src/app/router/mod.rs

@@ -79,7 +79,7 @@ impl Router {
         for r in self.rules.iter() {
             if sess.destination.is_domain() && r.should_resolve_ip() && !sess_resolved {
                 debug!(
-                    "rule {r} local resolving domain {}",
+                    "rule `{r}` resolving domain {} locally",
                     sess.destination.domain().unwrap()
                 );
                 if let Ok(Some(ip)) = self

clash_lib/src/app/router/rules/geoip.rs

@@ -16,7 +16,7 @@ pub struct GeoIP {
 
 impl std::fmt::Display for GeoIP {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{} geoip {}", self.target, self.country_code)
+        write!(f, "GeoIP({} - {})", self.target, self.country_code)
     }
 }
 

clash_lib/src/config/internal/config.rs

@@ -111,7 +111,8 @@ impl TryFrom<def::Config> for Config {
                 .rule_provider
                 .map(|m| {
                     m.into_iter()
-                        .try_fold(HashMap::new(), |mut rv, (name, body)| {
+                        .try_fold(HashMap::new(), |mut rv, (name, mut body)| {
+                            body.insert("name".to_owned(), serde_yaml::Value::String(name.clone()));
                             let provider = RuleProviderDef::try_from(body).map_err(|x| {
                                 Error::InvalidConfig(format!(
                                     "invalid rule provider {}: {}",
@@ -186,7 +187,8 @@ impl TryFrom<def::Config> for Config {
                 .proxy_provider
                 .map(|m| {
                     m.into_iter()
-                        .try_fold(HashMap::new(), |mut rv, (name, body)| {
+                        .try_fold(HashMap::new(), |mut rv, (name, mut body)| {
+                            body.insert("name".to_owned(), serde_yaml::Value::String(name.clone()));
                             let provider =
                                 OutboundProxyProviderDef::try_from(body).map_err(|x| {
                                     Error::InvalidConfig(format!(

clash_lib/src/proxy/converters/vmess.rs

@@ -68,7 +68,11 @@ impl TryFrom<&OutboundVmess> for AnyOutboundHandler {
                         .as_ref()
                         .map(|x| {
                             VmessTransport::H2(Http2Option {
-                                host: x.host.as_ref().map(|x| x.to_owned()).unwrap_or_default(),
+                                host: x
+                                    .host
+                                    .as_ref()
+                                    .map(|x| x.to_owned())
+                                    .unwrap_or(vec![s.server.to_owned()]),
                                 path: x.path.as_ref().map(|x| x.to_owned()).unwrap_or_default(),
                             })
                         })

clash_lib/src/proxy/direct/mod.rs

@@ -6,14 +6,15 @@ use crate::app::dns::ThreadSafeDNSResolver;
 use crate::config::internal::proxy::PROXY_DIRECT;
 use crate::proxy::datagram::OutboundDatagramImpl;
 use crate::proxy::utils::{new_tcp_stream, new_udp_socket};
-use crate::proxy::{AnyOutboundHandler, AnyStream, OutboundHandler};
-use crate::session::{Session, SocksAddr};
+use crate::proxy::{AnyOutboundHandler, OutboundHandler};
+use crate::session::Session;
 
 use async_trait::async_trait;
 use serde::Serialize;
 use std::sync::Arc;
 
-use super::OutboundType;
+use super::utils::RemoteConnector;
+use super::{ConnectorType, OutboundType};
 
 #[derive(Serialize)]
 pub struct Handler;
@@ -35,10 +36,6 @@ impl OutboundHandler for Handler {
         OutboundType::Direct
     }
 
-    async fn remote_addr(&self) -> Option<SocksAddr> {
-        None
-    }
-
     async fn support_udp(&self) -> bool {
         true
     }
@@ -63,15 +60,6 @@ impl OutboundHandler for Handler {
         Ok(Box::new(s))
     }
 
-    async fn proxy_stream(
-        &self,
-        s: AnyStream,
-        #[allow(unused_variables)] sess: &Session,
-        #[allow(unused_variables)] _resolver: ThreadSafeDNSResolver,
-    ) -> std::io::Result<AnyStream> {
-        Ok(s)
-    }
-
     async fn connect_datagram(
         &self,
         sess: &Session,
@@ -90,4 +78,50 @@ impl OutboundHandler for Handler {
         d.append_to_chain(self.name()).await;
         Ok(Box::new(d))
     }
+
+    async fn support_connector(&self) -> ConnectorType {
+        ConnectorType::Tcp
+    }
+
+    async fn connect_stream_with_connector(
+        &self,
+        sess: &Session,
+        resolver: ThreadSafeDNSResolver,
+        connector: &dyn RemoteConnector,
+    ) -> std::io::Result<BoxedChainedStream> {
+        let s = connector
+            .connect_stream(
+                resolver,
+                sess.destination.host().as_str(),
+                sess.destination.port(),
+                None,
+                #[cfg(any(target_os = "linux", target_os = "android"))]
+                None,
+            )
+            .await?;
+        let s = ChainedStreamWrapper::new(s);
+        s.append_to_chain(self.name()).await;
+        Ok(Box::new(s))
+    }
+
+    async fn connect_datagram_with_connector(
+        &self,
+        sess: &Session,
+        resolver: ThreadSafeDNSResolver,
+        connector: &dyn RemoteConnector,
+    ) -> std::io::Result<BoxedChainedDatagram> {
+        let d = connector
+            .connect_datagram(
+                resolver,
+                None,
+                &sess.destination,
+                sess.iface.as_ref(),
+                #[cfg(any(target_os = "linux", target_os = "android"))]
+                None,
+            )
+            .await?;
+        let d = ChainedDatagramWrapper::new(d);
+        d.append_to_chain(self.name()).await;
+        Ok(Box::new(d))
+    }
 }