add logs api

.gitignore

@@ -25,3 +25,6 @@ venv/
 /bazel-*
 # Ignore outputs generated during Bazel bootstrapping.
 /output/
+
+# don't check in this real config
+ignore.yaml

Cargo.Bazel.lock

@@ -1,5 +1,5 @@
 {
-  "checksum": "4c7d42924b2314ce069523e5915170a57d8303ede8bae1fcc0f5cbcfcb256e1a",
+  "checksum": "eb54538b6c466464ff68b5bebe68dc857d719e9b327fcc10cc20a4e90d556649",
   "crates": {
     "addr2line 0.20.0": {
       "name": "addr2line",
@@ -1294,7 +1294,8 @@
             "original-uri",
             "query",
             "tokio",
-            "tower-log"
+            "tower-log",
+            "ws"
           ],
           "selects": {}
         },
@@ -1308,6 +1309,10 @@
               "id": "axum-core 0.3.4",
               "target": "axum_core"
             },
+            {
+              "id": "base64 0.21.2",
+              "target": "base64"
+            },
             {
               "id": "bitflags 1.3.2",
               "target": "bitflags"
@@ -1372,6 +1377,10 @@
               "id": "serde_urlencoded 0.7.1",
               "target": "serde_urlencoded"
             },
+            {
+              "id": "sha1 0.10.5",
+              "target": "sha1"
+            },
             {
               "id": "sync_wrapper 0.1.2",
               "target": "sync_wrapper"
@@ -1380,6 +1389,10 @@
               "id": "tokio 1.29.1",
               "target": "tokio"
             },
+            {
+              "id": "tokio-tungstenite 0.20.0",
+              "target": "tokio_tungstenite"
+            },
             {
               "id": "tower 0.4.13",
               "target": "tower"
@@ -3431,6 +3444,10 @@
               "id": "trust-dns-proto 0.22.0",
               "target": "trust_dns_proto"
             },
+            {
+              "id": "trust-dns-resolver 0.22.0",
+              "target": "trust_dns_resolver"
+            },
             {
               "id": "url 2.4.0",
               "target": "url"

README.md

@@ -2,7 +2,8 @@
 Just a toy for fun, don't use please :+|
 
 TODOs
-- [ ] proxy rules (relay, select, ...)
+- [x] proxy rules (relay, select, ...)
 - [ ] dashboard
 - [ ] DNS server
-- [ ] trojan proto
+- [ ] trojan proto
+- [ ] hc and fetchers

clash_lib/Cargo.toml

@@ -45,13 +45,14 @@ md-5 = "0.10.5"
 chacha20poly1305 = "0.10"
 aes-gcm = "0.10"
 filetime = "0.2"
-axum = "0.6.20"
+axum = { version = "0.6.20", features = ["ws"] }
 
 
 serde = { version = "1.0", features=["derive"] }
 serde_yaml = "0.9"
 
 trust-dns-client = "0.22"
+trust-dns-resolver = "0.22.0"
 trust-dns-proto = { version = "0.22", features = ["dns-over-rustls", "dns-over-https-rustls"]}
 # DoH
 rustls = { version  = "0.20", features=["dangerous_configuration"] }

clash_lib/src/app/api/handlers/hello.rs

@@ -0,0 +1,9 @@
+use std::collections::HashMap;
+
+use axum::response::IntoResponse;
+
+pub async fn handle() -> axum::response::Response {
+    let mut val = HashMap::new();
+    val.insert("hello".to_owned(), "clash-rs".to_owned());
+    axum::response::Json(val).into_response()
+}

clash_lib/src/app/api/handlers/log.rs

@@ -0,0 +1,30 @@
+use std::{net::SocketAddr, sync::Arc};
+
+use axum::{
+    extract::{ws::Message, ConnectInfo, State, WebSocketUpgrade},
+    response::IntoResponse,
+};
+
+use tracing::{debug, warn};
+
+use crate::app::api::AppState;
+
+pub async fn handle(
+    ws: WebSocketUpgrade,
+    ConnectInfo(addr): ConnectInfo<SocketAddr>,
+    State(state): State<Arc<AppState>>,
+) -> impl IntoResponse {
+    debug!("ws connect from {}", addr);
+    ws.on_failed_upgrade(|e| {
+        warn!("ws upgrade error: {}", e);
+    })
+    .on_upgrade(move |mut socket| async move {
+        let mut rx = state.log_source_tx.subscribe();
+        while let Ok(msg) = rx.recv().await {
+            if let Err(e) = socket.send(Message::Text(msg)).await {
+                warn!("ws send error: {}", e);
+                break;
+            }
+        }
+    })
+}

clash_lib/src/app/api/handlers/mod.rs

@@ -0,0 +1,2 @@
+pub mod hello;
+pub mod log;

clash_lib/src/app/api/mod.rs

@@ -1,20 +1,32 @@
+use std::{net::SocketAddr, sync::Arc};
+
 use axum::{routing::get, Router};
+
+use tokio::sync::broadcast::Sender;
 use tracing::info;
 
 use crate::{config::internal::config::Controller, Runner};
 
-async fn root() -> &'static str {
-    "Hello, World!"
+mod handlers;
+
+pub struct AppState {
+    log_source_tx: Sender<String>,
 }
 
-pub fn get_api_runner(controller_cfg: Controller) -> Option<Runner> {
+pub fn get_api_runner(controller_cfg: Controller, log_source: Sender<String>) -> Option<Runner> {
     if let Some(bind_addr) = controller_cfg.external_controller {
+        let app_state = AppState {
+            log_source_tx: log_source,
+        };
         let addr = bind_addr.parse().unwrap();
         let runner = async move {
             info!("Starting API server at {}", addr);
-            let app = Router::new().route("/", get(root));
+            let app = Router::new()
+                .route("/", get(handlers::hello::handle))
+                .route("/logs", get(handlers::log::handle))
+                .with_state(Arc::new(app_state));
             axum::Server::bind(&addr)
-                .serve(app.into_make_service())
+                .serve(app.into_make_service_with_connect_info::<SocketAddr>())
                 .await
                 .unwrap();
         };

clash_lib/src/app/dns/mod.rs

@@ -14,6 +14,7 @@ mod fakeip;
 mod filters;
 mod helper;
 pub mod resolver;
+mod system;
 
 use crate::dns::dns_client::DNSNetMode;
 

clash_lib/src/app/dns/resolver.rs

@@ -11,15 +11,16 @@ use trust_dns_proto::{op, rr};
 #[cfg(test)]
 use mockall::automock;
 
+use crate::app::ThreadSafeDNSResolver;
 use crate::dns::helper::make_clients;
 use crate::dns::ThreadSafeDNSClient;
 use crate::{common::trie, Error};
 
+use super::system::SystemResolver;
 use super::{
     filters::{DomainFilter, FallbackDomainFilter, FallbackIPFilter, GeoIPFilter, IPNetFilter},
     Config,
 };
-use super::{DNSNetMode, NameServer};
 
 static TTL: Duration = Duration::from_secs(60);
 
@@ -267,7 +268,10 @@ impl ClashResolver for Resolver {
 
 impl Resolver {
     /// For testing purpose
+    #[cfg(test)]
     pub async fn new_default() -> Self {
+        use super::{DNSNetMode, NameServer};
+
         Resolver {
             ipv6: false,
             hosts: None,
@@ -288,7 +292,11 @@ impl Resolver {
         }
     }
 
-    pub async fn new(cfg: Config) -> Self {
+    pub async fn new(cfg: Config) -> ThreadSafeDNSResolver {
+        if !cfg.enable {
+            return Arc::new(SystemResolver::new().expect("failed to create system resolver"));
+        }
+
         let default_resolver = Arc::new(Resolver {
             ipv6: false,
             hosts: None,
@@ -356,7 +364,7 @@ impl Resolver {
             },
         };
 
-        r
+        Arc::new(r)
     }
 
     pub async fn batch_exchange(
@@ -365,7 +373,6 @@ impl Resolver {
     ) -> anyhow::Result<op::Message> {
         let mut queries = Vec::new();
         for c in clients {
-            // TODO: how to use .map()
             queries.push(
                 async move {
                     c.lock()

clash_lib/src/app/dns/system.rs

@@ -0,0 +1,44 @@
+use async_trait::async_trait;
+use rand::seq::IteratorRandom;
+use trust_dns_resolver::TokioAsyncResolver;
+
+use super::ClashResolver;
+
+pub(crate) struct SystemResolver {
+    resolver: TokioAsyncResolver,
+}
+
+impl SystemResolver {
+    pub fn new() -> anyhow::Result<Self> {
+        let resolver = TokioAsyncResolver::tokio_from_system_conf()?;
+        Ok(Self { resolver })
+    }
+}
+
+#[async_trait]
+impl ClashResolver for SystemResolver {
+    async fn resolve(&self, host: &str) -> anyhow::Result<Option<std::net::IpAddr>> {
+        let response = self.resolver.lookup_ip(host).await?;
+        Ok(response.iter().choose(&mut rand::thread_rng()))
+    }
+    async fn resolve_v4(&self, host: &str) -> anyhow::Result<Option<std::net::Ipv4Addr>> {
+        let response = self.resolver.lookup_ip(host).await?;
+        Ok(response
+            .iter()
+            .filter_map(|ip| match ip {
+                std::net::IpAddr::V4(ip) => Some(ip),
+                _ => None,
+            })
+            .choose(&mut rand::thread_rng()))
+    }
+    async fn resolve_v6(&self, host: &str) -> anyhow::Result<Option<std::net::Ipv6Addr>> {
+        let response = self.resolver.lookup_ip(host).await?;
+        Ok(response
+            .iter()
+            .filter_map(|ip| match ip {
+                std::net::IpAddr::V6(ip) => Some(ip),
+                _ => None,
+            })
+            .choose(&mut rand::thread_rng()))
+    }
+}

clash_lib/src/app/logging.rs

@@ -1,6 +1,9 @@
 use crate::def::LogLevel;
+use tokio::sync::broadcast::Sender;
+use tracing::debug;
 use tracing_subscriber::filter::Targets;
 use tracing_subscriber::prelude::*;
+use tracing_subscriber::Layer;
 use tracing_subscriber::{filter, EnvFilter};
 
 impl From<LogLevel> for filter::LevelFilter {
@@ -15,14 +18,46 @@ impl From<LogLevel> for filter::LevelFilter {
     }
 }
 
-pub fn setup_logging(level: LogLevel) -> anyhow::Result<()> {
+pub struct EventCollector(Vec<Sender<String>>);
+
+impl EventCollector {
+    pub fn new(recivers: Vec<Sender<String>>) -> Self {
+        Self(recivers)
+    }
+}
+
+impl<S> Layer<S> for EventCollector
+where
+    S: tracing::Subscriber,
+{
+    fn on_event(
+        &self,
+        event: &tracing::Event<'_>,
+        _ctx: tracing_subscriber::layer::Context<'_, S>,
+    ) {
+        let mut msg = vec![];
+        msg.push(format!("{}", event.metadata().level()));
+        msg.push(format!("{}", event.metadata().target()));
+        msg.push(format!("{}", event.metadata().name()));
+        for field in event.fields() {
+            msg.push(format!("{}", field.name()));
+        }
+
+        for tx in &self.0 {
+            _ = tx.send(msg.join(""));
+        }
+    }
+}
+
+pub fn setup_logging(level: LogLevel, collector: EventCollector) -> anyhow::Result<()> {
     let filter = EnvFilter::builder()
         .with_default_directive(filter::LevelFilter::from(level).into())
         .from_env_lossy();
 
     let subscriber = tracing_subscriber::registry()
         .with(filter)
         .with(Targets::new().with_target("clash", level))
+        .with(collector)
         .with(
             tracing_subscriber::fmt::Layer::new()
                 .with_ansi(atty::is(atty::Stream::Stdout))

clash_lib/src/config/def.rs

@@ -11,9 +11,12 @@ use serde_yaml::Value;
 #[derive(Serialize, Deserialize, Default, Copy, Clone)]
 #[serde(rename_all = "lowercase")]
 pub enum RunMode {
+    #[serde(alias = "Global")]
     Global,
     #[default]
+    #[serde(alias = "Rule")]
     Rule,
+    #[serde(alias = "Direct")]
     Direct,
 }
 

clash_lib/src/config/internal/proxy.rs

@@ -156,6 +156,7 @@ pub struct OutboundVmess {
     pub skip_cert_verify: Option<bool>,
     pub server_name: Option<String>,
     pub network: Option<String>,
+    #[serde(alias = "ws-opts")]
     pub ws_opts: Option<WsOpt>,
 }