diff --git a/Cargo.lock b/Cargo.lock index a718ab3..c8c87a4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -76,12 +76,27 @@ version = "2.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "bumpalo" version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + [[package]] name = "bytes" version = "1.12.0" @@ -169,6 +184,41 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + +[[package]] +name = "data-encoding" +version = "2.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" + +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "displaydoc" version = "0.2.6" @@ -198,7 +248,13 @@ dependencies = [ "ipnet", "rand 0.8.6", "reqwest", + "rustls", "tokio", + "tokio-rustls", + "tokio-tungstenite", + "url", + "uuid", + "webpki-roots 0.26.11", ] [[package]] @@ -304,6 +360,16 @@ dependencies = [ "slab", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "getrandom" version = "0.2.17" @@ -409,7 +475,7 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", - "webpki-roots", + "webpki-roots 1.0.8", ] [[package]] @@ -698,7 +764,7 @@ dependencies = [ "rustc-hash", "rustls", "socket2", - "thiserror", + "thiserror 2.0.18", "tokio", "tracing", "web-time", @@ -719,7 +785,7 @@ dependencies = [ "rustls", "rustls-pki-types", "slab", - "thiserror", + "thiserror 2.0.18", "tinyvec", "tracing", "web-time", @@ -848,7 +914,7 @@ dependencies = [ "wasm-bindgen", "wasm-bindgen-futures", "web-sys", - "webpki-roots", + "webpki-roots 1.0.8", ] [[package]] @@ -877,6 +943,7 @@ version = "0.23.41" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b92b125634d9b795e7beca796cc790df15a7fb38323bf3196fda83292d06b1f" dependencies = [ + "log", "once_cell", "ring", "rustls-pki-types", @@ -972,6 +1039,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha1" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "shlex" version = "2.0.1" @@ -1049,13 +1127,33 @@ dependencies = [ "syn", ] +[[package]] +name = "thiserror" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" +dependencies = [ + "thiserror-impl 1.0.69", +] + [[package]] name = "thiserror" version = "2.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" dependencies = [ - "thiserror-impl", + "thiserror-impl 2.0.18", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn", ] [[package]] @@ -1130,6 +1228,18 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite", +] + [[package]] name = "tower" version = "0.5.3" @@ -1200,6 +1310,30 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18e5b8366ee7a95b16d32197d0b2604b43a0be89dc5fac9f8e96ccafbaedda8a" +dependencies = [ + "byteorder", + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.8.6", + "sha1", + "thiserror 1.0.69", + "utf-8", +] + +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "unicode-ident" version = "1.0.24" @@ -1230,6 +1364,12 @@ dependencies = [ "serde", ] +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "utf8_iter" version = "1.0.4" @@ -1242,6 +1382,22 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "uuid" +version = "1.23.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "144d6b123cef80b301b8f72a9e2ca4370ddec21950d0a103dd22c437006d2db7" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + [[package]] name = "want" version = "0.3.1" @@ -1341,6 +1497,15 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.8", +] + [[package]] name = "webpki-roots" version = "1.0.8" diff --git a/Cargo.toml b/Cargo.toml index 04a51ea..7e855ad 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,7 +14,13 @@ indicatif = "0.17" ipnet = "2" rand = "0.8" reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] } -tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "time"] } +rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12", "logging"] } +tokio = { version = "1", features = ["io-util", "macros", "net", "rt-multi-thread", "time"] } +tokio-rustls = { version = "0.26", default-features = false, features = ["ring", "tls12", "logging"] } +tokio-tungstenite = "0.24" +url = "2" +uuid = "1" +webpki-roots = "0.26" [profile.release] opt-level = 3 diff --git a/src/latency.rs b/src/latency.rs new file mode 100644 index 0000000..34a7203 --- /dev/null +++ b/src/latency.rs @@ -0,0 +1,108 @@ +//! Stage 2 (`latency`): measure real proxy latency for each candidate IP by +//! tunnelling an HTTPS request through the node and timing it. + +use std::net::IpAddr; +use std::time::{Duration, Instant}; + +use anyhow::{Result, anyhow}; +use futures::stream::{self, StreamExt}; +use rustls::pki_types::ServerName; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +use crate::vless::{self, VlessNode}; + +/// Latency probe target: a tiny 204 endpoint, like the Python version uses. +const TARGET_HOST: &str = "www.google.com"; +const TARGET_PORT: u16 = 443; +const TARGET_PATH: &str = "/generate_204"; + +/// One IP and its measured real round-trip latency. +#[derive(Clone, Debug)] +pub struct LatencyResult { + pub ip: IpAddr, + pub latency: Duration, +} + +/// Progress event, emitted once per finished measurement. +pub struct Probed { + pub ip: IpAddr, + pub latency: Option, + pub done: usize, + #[allow(dead_code)] // available to callers that want a denominator. + pub total: usize, +} + +/// Measure latency for all `ips` with bounded concurrency, calling +/// `on_progress` after each. Returns successes sorted fastest first. +pub async fn run( + node: &VlessNode, + ips: &[IpAddr], + concurrency: usize, + timeout: Duration, + mut on_progress: impl FnMut(Probed), +) -> Vec { + let total = ips.len(); + let mut done = 0usize; + let mut results: Vec = Vec::new(); + + let mut stream = stream::iter(ips.iter().copied().map(|ip| async move { + (ip, measure(node, ip, timeout).await) + })) + .buffer_unordered(concurrency.max(1)); + + while let Some((ip, outcome)) = stream.next().await { + done += 1; + let latency = outcome.ok(); + if let Some(d) = latency { + results.push(LatencyResult { ip, latency: d }); + } + on_progress(Probed { ip, latency, done, total }); + } + + results.sort_by_key(|r| r.latency); + results +} + +/// Tunnel a single `GET /generate_204` through the node via `ip` and return +/// the total time to a 2xx/3xx response. +pub async fn measure(node: &VlessNode, ip: IpAddr, timeout: Duration) -> Result { + let started = Instant::now(); + let attempt = tokio::time::timeout(timeout, async { + let tunnel = vless::connect(node, ip, TARGET_HOST, TARGET_PORT).await?; + + // Inner TLS to the real target, through the tunnel (TLS-in-WS-in-TLS). + let sni = ServerName::try_from(TARGET_HOST.to_string())?; + let mut stream = vless::tls_connector().connect(sni, tunnel).await?; + + let request = format!( + "GET {TARGET_PATH} HTTP/1.1\r\nHost: {TARGET_HOST}\r\n\ + User-Agent: fast-xray\r\nAccept: */*\r\nConnection: close\r\n\r\n" + ); + stream.write_all(request.as_bytes()).await?; + stream.flush().await?; + + let mut buf = [0u8; 4096]; + let n = stream.read(&mut buf).await?; + if n == 0 { + return Err(anyhow!("empty response")); + } + let head = String::from_utf8_lossy(&buf[..n]); + let status = head + .lines() + .next() + .and_then(|line| line.split_whitespace().nth(1)) + .and_then(|code| code.parse::().ok()) + .ok_or_else(|| anyhow!("no http status"))?; + if !(200..400).contains(&status) { + return Err(anyhow!("http status {status}")); + } + Ok(()) + }) + .await; + + match attempt { + Ok(Ok(())) => Ok(started.elapsed()), + Ok(Err(e)) => Err(e), + Err(_) => Err(anyhow!("timeout")), + } +} diff --git a/src/main.rs b/src/main.rs index 078bb38..865950b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,18 +4,47 @@ //! ping — collect the fastest reachable Cloudflare IPs over TCP. mod cloudflare; +mod latency; mod ping; +mod vless; +use std::net::IpAddr; use std::path::{Path, PathBuf}; use std::time::Duration; -use anyhow::{Context, Result}; +use anyhow::{Context, Result, anyhow}; use clap::{Args, Parser, Subcommand}; use console::style; use indicatif::{ProgressBar, ProgressStyle}; use cloudflare::Family; -use ping::{PingResult, Progress}; +use ping::Progress; +use vless::VlessNode; + +/// Anything with an IP and a measured latency, so the table/file writers work +/// for both the ping and latency stages. +trait Ranked { + fn ip(&self) -> IpAddr; + fn latency(&self) -> Duration; +} + +impl Ranked for ping::PingResult { + fn ip(&self) -> IpAddr { + self.ip + } + fn latency(&self) -> Duration { + self.latency + } +} + +impl Ranked for latency::LatencyResult { + fn ip(&self) -> IpAddr { + self.ip + } + fn latency(&self) -> Duration { + self.latency + } +} #[derive(Parser)] #[command(name = "fast-xray", about = "Fast Cloudflare IP optimizer for CDN-fronted Xray/VLESS nodes.")] @@ -28,6 +57,39 @@ struct Cli { enum Command { /// Probe Cloudflare IPs over TCP 443 and collect the fastest reachable ones. Ping(PingArgs), + /// Measure real proxy latency for candidate IPs through the node. + Latency(LatencyArgs), +} + +#[derive(Args)] +struct LatencyArgs { + /// Input vless:// node URL (or use --node-file). + node: Option, + + /// Read the vless:// node URL from a file (avoids shell escaping of `&`). + #[arg(long)] + node_file: Option, + + /// IP list to test (one per line), e.g. the ping stage output. + #[arg(short = 'i', long, default_value = "result/ip.txt")] + input: PathBuf, + + /// Concurrent tunnels. + #[arg(short = 'c', long, default_value_t = 50)] + concurrency: usize, + + /// Per-IP timeout in seconds for the whole tunnel + request. + #[arg(short = 't', long, default_value_t = 5.0)] + timeout: f64, + + /// Output directory. Writes /latency.csv (IP + latency) and + /// /latency.txt (plain IPs, fastest first). + #[arg(short = 'o', long, default_value = "result")] + output: PathBuf, + + /// Print the full result table at the end. + #[arg(short = 'v', long)] + verbose: bool, } #[derive(Args)] @@ -60,8 +122,8 @@ struct PingArgs { #[arg(long)] max_probe: Option, - /// Write the resulting IPs (one per line, fastest first) to this file. - #[arg(short = 'o', long, default_value = "result/ip.txt")] + /// Output directory. Writes /ip.txt (plain IPs, fastest first). + #[arg(short = 'o', long, default_value = "result")] output: PathBuf, /// Print the full result table (IP + latency) at the end. @@ -71,9 +133,13 @@ struct PingArgs { #[tokio::main] async fn main() -> Result<()> { + // Process-wide TLS crypto provider for our own rustls client configs. + let _ = rustls::crypto::ring::default_provider().install_default(); + let cli = Cli::parse(); match cli.command { Command::Ping(args) => run_ping(args).await, + Command::Latency(args) => run_latency(args).await, } } @@ -139,12 +205,88 @@ async fn run_ping(args: PingArgs) -> Result<()> { if args.verbose { print_table(&results); } - write_output(&results, &args.output)?; + write_ips(&results, &args.output.join("ip.txt"))?; Ok(()) } +async fn run_latency(args: LatencyArgs) -> Result<()> { + // Resolve the node from --node-file or the positional argument. + let node_text = match (&args.node_file, &args.node) { + (Some(path), _) => std::fs::read_to_string(path) + .with_context(|| format!("read node file {}", path.display()))? + .trim() + .trim_start_matches('\u{feff}') + .to_string(), + (None, Some(node)) => node.clone(), + (None, None) => return Err(anyhow!("provide a vless:// node or --node-file")), + }; + let node = VlessNode::parse(&node_text)?; + + let ips = read_ip_list(&args.input)?; + if ips.is_empty() { + return Err(anyhow!("no valid IPs in {}", args.input.display())); + } + eprintln!( + "Node host: {} path: {} | testing {} IPs", + style(&node.host).cyan(), + style(&node.path).cyan(), + ips.len() + ); + + let bar = ProgressBar::new(ips.len() as u64); + bar.set_style( + ProgressStyle::with_template( + "{spinner:.cyan} [{bar:32.green/dim}] {pos}/{len} {msg}", + ) + .unwrap() + .progress_chars("=>-"), + ); + bar.enable_steady_tick(Duration::from_millis(90)); + + let results = latency::run( + &node, + &ips, + args.concurrency.max(1), + Duration::from_secs_f64(args.timeout), + |p| { + bar.set_position(p.done as u64); + match p.latency { + Some(d) => { + let ms = d.as_secs_f64() * 1000.0; + bar.set_message(format!("{} {} {:.0}ms", style("✓").green(), p.ip, ms)); + } + None => bar.set_message(format!("{} {}", style("·").dim(), p.ip)), + } + }, + ) + .await; + bar.finish_and_clear(); + + if results.is_empty() { + eprintln!("{} no IP passed the real latency test", style("✗").red().bold()); + return Ok(()); + } + + if args.verbose { + print_table(&results); + } + write_csv(&results, &args.output.join("latency.csv"))?; + write_ips(&results, &args.output.join("latency.txt"))?; + Ok(()) +} + +/// Read an IP-per-line file, skipping blanks and unparseable lines. +fn read_ip_list(path: &Path) -> Result> { + let text = std::fs::read_to_string(path) + .with_context(|| format!("read IP list {}", path.display()))?; + Ok(text + .lines() + .filter_map(|line| line.trim().parse::().ok()) + .collect()) +} + /// Pretty, colored result table on stderr (human-facing). -fn print_table(results: &[PingResult]) { +fn print_table(results: &[T]) { eprintln!(); eprintln!( "{}", @@ -152,7 +294,7 @@ fn print_table(results: &[PingResult]) { ); eprintln!(" {:>3} {:<39} {:>9}", "#", "IP", "Latency"); for (i, r) in results.iter().enumerate() { - let ms = r.latency.as_secs_f64() * 1000.0; + let ms = r.latency().as_secs_f64() * 1000.0; let latency = format!("{ms:>6.1} ms"); let colored = if ms < 150.0 { style(latency).green() @@ -161,23 +303,19 @@ fn print_table(results: &[PingResult]) { } else { style(latency).red() }; - eprintln!(" {:>3} {:<39} {}", i + 1, r.ip.to_string(), colored); + eprintln!(" {:>3} {:<39} {}", i + 1, r.ip().to_string(), colored); } eprintln!(); } /// Write IPs only (one per line, fastest first) to `path`. -fn write_output(results: &[PingResult], path: &Path) -> Result<()> { +fn write_ips(results: &[T], path: &Path) -> Result<()> { let mut body = String::with_capacity(results.len() * 16); for r in results { - body.push_str(&r.ip.to_string()); + body.push_str(&r.ip().to_string()); body.push('\n'); } - if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) { - std::fs::create_dir_all(parent) - .with_context(|| format!("create dir for {}", path.display()))?; - } - std::fs::write(path, &body).with_context(|| format!("write {}", path.display()))?; + write_file(path, &body)?; eprintln!( "{} saved {} IPs -> {}", style("✓").green().bold(), @@ -186,3 +324,27 @@ fn write_output(results: &[PingResult], path: &Path) -> Result<()> { ); Ok(()) } + +/// Write an IP + latency CSV, fastest first. +fn write_csv(results: &[T], path: &Path) -> Result<()> { + let mut body = String::from("IP,Latency(ms)\n"); + for r in results { + body.push_str(&format!("{},{:.2}\n", r.ip(), r.latency().as_secs_f64() * 1000.0)); + } + write_file(path, &body)?; + eprintln!( + "{} saved CSV -> {}", + style("✓").green().bold(), + style(path.display()).cyan() + ); + Ok(()) +} + +/// Write `body` to `path`, creating parent directories as needed. +fn write_file(path: &Path, body: &str) -> Result<()> { + if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) { + std::fs::create_dir_all(parent) + .with_context(|| format!("create dir for {}", path.display()))?; + } + std::fs::write(path, body).with_context(|| format!("write {}", path.display())) +} diff --git a/src/vless.rs b/src/vless.rs new file mode 100644 index 0000000..be87f75 --- /dev/null +++ b/src/vless.rs @@ -0,0 +1,377 @@ +//! Minimal VLESS + WebSocket + TLS outbound, just enough to tunnel one TCP +//! stream through a Cloudflare-fronted node — no external xray needed. +//! +//! Only the simplest node shape is supported (the common CF-CDN case): +//! `encryption=none`, no flow / no XTLS-Vision, transport `ws`, security `tls`. +//! Under those settings the VLESS body is a raw byte stream in both directions +//! (confirmed against xray-core `encoding.go` / `addons.go`). +//! +//! Layering, outermost to innermost: +//! ```text +//! TCP(cf_ip:443) +//! └─ outer TLS (SNI = node host) tokio-rustls +//! └─ WebSocket (Host = node host, path) tokio-tungstenite +//! └─ VLESS header (uuid + target) this module +//! └─ raw stream to target host:port ← what connect() returns +//! ``` + +use std::io; +use std::net::IpAddr; +use std::pin::Pin; +use std::sync::{Arc, OnceLock}; +use std::task::{Context, Poll}; + +use anyhow::{Context as _, Result, anyhow}; +use futures::sink::Sink; +use futures::stream::Stream; +use rustls::RootCertStore; +use rustls::pki_types::ServerName; +use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, ReadBuf}; +use tokio::net::TcpStream; +use tokio_rustls::TlsConnector; +use tokio_rustls::client::TlsStream; +use tokio_tungstenite::WebSocketStream; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use url::Url; +use uuid::Uuid; + +const VLESS_VERSION: u8 = 0; +const CMD_TCP: u8 = 1; +const ADDR_IPV4: u8 = 1; +const ADDR_DOMAIN: u8 = 2; +const ADDR_IPV6: u8 = 3; + +/// A parsed VLESS node. `host` is the original domain — it stays the TLS SNI +/// and WebSocket Host even when we connect to a different Cloudflare IP. +#[derive(Clone, Debug)] +#[allow(dead_code)] // network/encryption/remark are used by the later speed stage. +pub struct VlessNode { + pub uuid: String, + pub host: String, + pub port: u16, + pub path: String, + pub network: String, + pub encryption: String, + pub remark: String, +} + +impl VlessNode { + /// Parse a `vless://uuid@host:port?...#remark` URL. + pub fn parse(raw: &str) -> Result { + let url = Url::parse(raw.trim()).context("parse vless url")?; + if url.scheme() != "vless" { + return Err(anyhow!("not a vless:// url")); + } + let uuid = url.username().to_string(); + if uuid.is_empty() { + return Err(anyhow!("vless url missing uuid")); + } + let url_host = url.host_str().ok_or_else(|| anyhow!("vless url missing host"))?; + + let mut host = url_host.to_string(); + let mut path = "/".to_string(); + let mut network = "ws".to_string(); + let mut encryption = "none".to_string(); + for (key, value) in url.query_pairs() { + match key.as_ref() { + // host/sni override the connect domain, like xray does. + "host" => host = value.into_owned(), + "sni" if host == url_host => host = value.into_owned(), + "path" => path = value.into_owned(), + "type" => network = value.into_owned(), + "encryption" => encryption = value.into_owned(), + _ => {} + } + } + + Ok(Self { + uuid, + host, + port: url.port().unwrap_or(443), + path, + network, + encryption, + remark: url.fragment().unwrap_or("").to_string(), + }) + } +} + +/// The raw byte stream to the target host, established through the node. +pub type VlessConn = VlessClientStream>>; + +/// Build a VLESS+WS+TLS tunnel through `cf_ip` and open a connection to +/// `target_host:target_port` inside it. The returned stream carries raw bytes +/// to the target (do your own TLS/HTTP on top). +pub async fn connect( + node: &VlessNode, + cf_ip: IpAddr, + target_host: &str, + target_port: u16, +) -> Result { + // 1. TCP to the Cloudflare edge IP. + let tcp = TcpStream::connect((cf_ip, node.port)).await.context("tcp connect")?; + tcp.set_nodelay(true).ok(); + + // 2. Outer TLS, SNI = original domain (this is what routes CF back to us). + let sni = ServerName::try_from(node.host.clone()).context("invalid sni host")?; + let tls = tls_connector().connect(sni, tcp).await.context("outer tls handshake")?; + + // 3. WebSocket upgrade. Using the domain in the URL makes tungstenite set + // Host = domain automatically; the path comes straight from the node. + let uri = format!("wss://{}{}", node.host, node.path); + let request = uri.into_client_request().context("build ws request")?; + let (ws, _resp) = tokio_tungstenite::client_async(request, tls) + .await + .context("websocket handshake")?; + let mut tunnel = WsByteStream::new(ws); + + // 4. VLESS request header (targets the inner destination), then flush so + // the server starts dialing the target before our payload arrives. + let header = build_request_header(node, target_host, target_port)?; + tunnel.write_all(&header).await.context("write vless header")?; + tunnel.flush().await.context("flush vless header")?; + + // 5. Strip the VLESS response header from the downlink lazily. + Ok(VlessClientStream::new(tunnel)) +} + +/// Encode the VLESS request header: version, uuid, no addons, TCP command, +/// then the target address/port. +fn build_request_header(node: &VlessNode, host: &str, port: u16) -> Result> { + let id = Uuid::parse_str(&node.uuid).context("invalid uuid")?; + let mut out = Vec::with_capacity(64); + out.push(VLESS_VERSION); + out.extend_from_slice(id.as_bytes()); + out.push(0); // addons length + out.push(CMD_TCP); + out.extend_from_slice(&port.to_be_bytes()); + match host.parse::() { + Ok(IpAddr::V4(v4)) => { + out.push(ADDR_IPV4); + out.extend_from_slice(&v4.octets()); + } + Ok(IpAddr::V6(v6)) => { + out.push(ADDR_IPV6); + out.extend_from_slice(&v6.octets()); + } + Err(_) => { + let bytes = host.as_bytes(); + if bytes.len() > 255 { + return Err(anyhow!("target host too long")); + } + out.push(ADDR_DOMAIN); + out.push(bytes.len() as u8); + out.extend_from_slice(bytes); + } + } + Ok(out) +} + +/// A process-wide TLS connector with the standard webpki roots and http/1.1 +/// ALPN. Reused for both the outer (to CF) and inner (to target) handshakes. +pub fn tls_connector() -> TlsConnector { + static CONNECTOR: OnceLock = OnceLock::new(); + CONNECTOR + .get_or_init(|| { + let mut roots = RootCertStore::empty(); + roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned()); + let mut config = rustls::ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth(); + config.alpn_protocols = vec![b"http/1.1".to_vec()]; + TlsConnector::from(Arc::new(config)) + }) + .clone() +} + +fn ws_err(e: tokio_tungstenite::tungstenite::Error) -> io::Error { + io::Error::new(io::ErrorKind::Other, e) +} + +/// Adapts a message-oriented WebSocket into a byte-oriented `AsyncRead`/ +/// `AsyncWrite`: every write becomes one binary frame; reads concatenate the +/// binary frames back into a byte stream. +pub struct WsByteStream { + inner: WebSocketStream, + rbuf: Vec, + rpos: usize, +} + +impl WsByteStream { + fn new(inner: WebSocketStream) -> Self { + Self { inner, rbuf: Vec::new(), rpos: 0 } + } +} + +impl AsyncRead for WsByteStream { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let me = self.get_mut(); + loop { + if me.rpos < me.rbuf.len() { + let n = (me.rbuf.len() - me.rpos).min(buf.remaining()); + buf.put_slice(&me.rbuf[me.rpos..me.rpos + n]); + me.rpos += n; + if me.rpos == me.rbuf.len() { + me.rbuf.clear(); + me.rpos = 0; + } + return Poll::Ready(Ok(())); + } + match Pin::new(&mut me.inner).poll_next(cx) { + Poll::Ready(Some(Ok(Message::Binary(data)))) => { + me.rbuf.extend_from_slice(data.as_ref()); + } + Poll::Ready(Some(Ok(Message::Close(_)))) | Poll::Ready(None) => { + return Poll::Ready(Ok(())); // EOF + } + Poll::Ready(Some(Ok(_))) => {} // ignore ping/pong/text, poll again + Poll::Ready(Some(Err(e))) => return Poll::Ready(Err(ws_err(e))), + Poll::Pending => return Poll::Pending, + } + } + } +} + +impl AsyncWrite for WsByteStream { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + let me = self.get_mut(); + match Pin::new(&mut me.inner).poll_ready(cx) { + Poll::Ready(Ok(())) => {} + Poll::Ready(Err(e)) => return Poll::Ready(Err(ws_err(e))), + Poll::Pending => return Poll::Pending, + } + match Pin::new(&mut me.inner).start_send(Message::Binary(buf.to_vec().into())) { + Ok(()) => Poll::Ready(Ok(buf.len())), + Err(e) => Poll::Ready(Err(ws_err(e))), + } + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let me = self.get_mut(); + Pin::new(&mut me.inner).poll_flush(cx).map_err(ws_err) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let me = self.get_mut(); + Pin::new(&mut me.inner).poll_close(cx).map_err(ws_err) + } +} + +/// Wraps the tunnel and transparently strips the VLESS response header +/// (`version`, `addons_len`, `addons…`) from the start of the downlink before +/// handing bytes to the caller. Writes pass straight through. +pub struct VlessClientStream { + inner: S, + phase: u8, // 0 = reading 2 header bytes, 1 = skipping addons, 2 = done + hdr_read: u8, + skip: usize, + leftover: Vec, + lo_pos: usize, +} + +impl VlessClientStream { + fn new(inner: S) -> Self { + Self { inner, phase: 0, hdr_read: 0, skip: 0, leftover: Vec::new(), lo_pos: 0 } + } + + /// Consume prelude bytes from `data`, returning how many were consumed. + fn consume_prelude(&mut self, data: &[u8]) -> usize { + let mut i = 0; + while i < data.len() && self.phase != 2 { + match self.phase { + 0 => { + if self.hdr_read == 0 { + self.hdr_read = 1; // version byte, value ignored + } else { + self.skip = data[i] as usize; // addons length + self.hdr_read = 2; + self.phase = if self.skip == 0 { 2 } else { 1 }; + } + i += 1; + } + 1 => { + let take = self.skip.min(data.len() - i); + self.skip -= take; + i += take; + if self.skip == 0 { + self.phase = 2; + } + } + _ => break, + } + } + i + } +} + +impl AsyncRead for VlessClientStream { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let me = self.get_mut(); + loop { + if me.lo_pos < me.leftover.len() { + let n = (me.leftover.len() - me.lo_pos).min(buf.remaining()); + buf.put_slice(&me.leftover[me.lo_pos..me.lo_pos + n]); + me.lo_pos += n; + if me.lo_pos == me.leftover.len() { + me.leftover.clear(); + me.lo_pos = 0; + } + return Poll::Ready(Ok(())); + } + if me.phase == 2 { + return Pin::new(&mut me.inner).poll_read(cx, buf); + } + // Still stripping the response header: read into scratch, drop the + // prelude, stash the remainder for the next loop iteration. + let mut scratch = [0u8; 2048]; + let mut rb = ReadBuf::new(&mut scratch); + match Pin::new(&mut me.inner).poll_read(cx, &mut rb) { + Poll::Ready(Ok(())) => {} + Poll::Ready(Err(e)) => return Poll::Ready(Err(e)), + Poll::Pending => return Poll::Pending, + } + let data = rb.filled(); + if data.is_empty() { + return Poll::Ready(Ok(())); // EOF before prelude completed + } + let consumed = me.consume_prelude(data); + if consumed < data.len() { + me.leftover.extend_from_slice(&data[consumed..]); + } + } + } +} + +impl AsyncWrite for VlessClientStream { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + let me = self.get_mut(); + Pin::new(&mut me.inner).poll_write(cx, buf) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let me = self.get_mut(); + Pin::new(&mut me.inner).poll_flush(cx) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let me = self.get_mut(); + Pin::new(&mut me.inner).poll_shutdown(cx) + } +}