From 426a40b2ef000257f3fab8394644db4c2df8967f Mon Sep 17 00:00:00 2001 From: chuan Date: Tue, 23 Jun 2026 20:15:02 +0800 Subject: [PATCH] feat: add speed measurement and export functionality --- src/main.rs | 280 ++++++++++++++++++++++++++++++++++++++++++++++++--- src/speed.rs | 158 +++++++++++++++++++++++++++++ src/vless.rs | 19 ++++ 3 files changed, 444 insertions(+), 13 deletions(-) create mode 100644 src/speed.rs diff --git a/src/main.rs b/src/main.rs index 865950b..7a58edc 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,6 +6,7 @@ mod cloudflare; mod latency; mod ping; +mod speed; mod vless; use std::net::IpAddr; @@ -46,6 +47,15 @@ impl Ranked for latency::LatencyResult { } } +impl Ranked for speed::SpeedResult { + 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.")] struct Cli { @@ -59,6 +69,66 @@ enum Command { Ping(PingArgs), /// Measure real proxy latency for candidate IPs through the node. Latency(LatencyArgs), + /// Download through the node to measure real speed. + Speed(SpeedArgs), + /// Build importable vless:// nodes from a valid-IP list and a node. + Export(ExportArgs), +} + +#[derive(Args)] +struct ExportArgs { + /// 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, + + /// Valid IP list: plain IPs, or a CSV whose first column is the IP. + #[arg(short = 'i', long, default_value = "result/ip.txt")] + input: PathBuf, + + /// Output directory. Writes /result.txt (importable nodes). + #[arg(short = 'o', long, default_value = "result")] + output: PathBuf, +} + +#[derive(Args)] +struct SpeedArgs { + /// 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, + + /// Input CSV from the latency stage (IP + latency). + #[arg(short = 'i', long, default_value = "result/latency.csv")] + input: PathBuf, + + /// Concurrent downloads. Default 1 for accurate throughput. + #[arg(short = 'c', long, default_value_t = 1)] + concurrency: usize, + + /// Per-IP download time cap in seconds. + #[arg(short = 't', long, default_value_t = 10.0)] + timeout: f64, + + /// Bytes to download per IP before stopping. + #[arg(long, default_value_t = 10_000_000)] + bytes: u64, + + /// Keep only the best N nodes. 0 means keep all. + #[arg(short = 'p', long, default_value_t = 0)] + top: usize, + + /// Output directory. Writes /result.txt (nodes) and /speed.csv. + #[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)] @@ -140,6 +210,8 @@ async fn main() -> Result<()> { match cli.command { Command::Ping(args) => run_ping(args).await, Command::Latency(args) => run_latency(args).await, + Command::Speed(args) => run_speed(args).await, + Command::Export(args) => run_export(args), } } @@ -210,17 +282,7 @@ async fn run_ping(args: PingArgs) -> Result<()> { } 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 node = resolve_node(&args.node_file, &args.node)?; let ips = read_ip_list(&args.input)?; if ips.is_empty() { @@ -275,13 +337,205 @@ async fn run_latency(args: LatencyArgs) -> Result<()> { Ok(()) } -/// Read an IP-per-line file, skipping blanks and unparseable lines. +async fn run_speed(args: SpeedArgs) -> Result<()> { + let node = resolve_node(&args.node_file, &args.node)?; + + let inputs = read_latency_csv(&args.input)?; + if inputs.is_empty() { + return Err(anyhow!("no IP+latency rows in {}", args.input.display())); + } + eprintln!( + "Node host: {} | speed-testing {} IPs", + style(&node.host).cyan(), + inputs.len() + ); + + let bar = ProgressBar::new(inputs.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 mut results = speed::run( + &node, + &inputs, + args.concurrency.max(1), + Duration::from_secs_f64(args.timeout), + args.bytes, + |p| { + bar.set_position(p.done as u64); + match p.speed_mbps { + Some(mbps) => { + bar.set_message(format!("{} {} {:.2}Mbps", style("✓").green(), p.ip, mbps)) + } + None => bar.set_message(format!("{} {}", style("·").dim(), p.ip)), + } + }, + ) + .await; + bar.finish_and_clear(); + + if results.is_empty() { + eprintln!("{} no IP passed the speed test", style("✗").red().bold()); + return Ok(()); + } + if args.top > 0 { + results.truncate(args.top); + } + + if args.verbose { + print_speed_table(&results); + } + write_speed_csv(&results, &args.output.join("speed.csv"))?; + write_ips(&results, &args.output.join("speed.txt"))?; + Ok(()) +} + +/// A parsed input row for export: IP plus whatever metrics the file carried. +struct ExportRow { + ip: IpAddr, + latency_ms: Option, + speed_mbps: Option, +} + +impl ExportRow { + /// Richest alias the available columns allow. + fn alias(&self) -> String { + match (self.speed_mbps, self.latency_ms) { + (Some(s), Some(l)) => format!("{s:.2}M-{l:.0}ms-{}", self.ip), + (_, Some(l)) => format!("{l:.0}ms-{}", self.ip), + _ => self.ip.to_string(), + } + } +} + +/// Build importable nodes from a valid-IP list + node. The input may be plain +/// IPs, latency CSV, or speed CSV — the alias adapts to whatever is present. +fn run_export(args: ExportArgs) -> Result<()> { + let node = resolve_node(&args.node_file, &args.node)?; + let rows = read_export_rows(&args.input)?; + if rows.is_empty() { + return Err(anyhow!("no valid IPs in {}", args.input.display())); + } + let path = args.output.join("result.txt"); + let mut body = String::new(); + for row in &rows { + body.push_str(&node.to_url(row.ip, &row.alias())); + body.push('\n'); + } + write_file(&path, &body)?; + eprintln!( + "{} built {} nodes -> {}", + style("✓").green().bold(), + rows.len(), + style(path.display()).cyan() + ); + Ok(()) +} + +/// Parse export input: first comma field is the IP; optional 2nd = latency(ms), +/// 3rd = speed(Mbps). Header/blank/unparseable lines are skipped. +fn read_export_rows(path: &Path) -> Result> { + let text = std::fs::read_to_string(path) + .with_context(|| format!("read input {}", path.display()))?; + let mut rows = Vec::new(); + for line in text.lines() { + let mut fields = line.split(','); + let Some(ip) = fields.next().and_then(|s| s.trim().parse::().ok()) else { + continue; + }; + rows.push(ExportRow { + ip, + latency_ms: fields.next().and_then(|s| s.trim().parse::().ok()), + speed_mbps: fields.next().and_then(|s| s.trim().parse::().ok()), + }); + } + Ok(rows) +} + +/// Resolve a node from --node-file (preferred) or the positional URL argument. +fn resolve_node(node_file: &Option, node: &Option) -> Result { + let text = match (node_file, 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")), + }; + VlessNode::parse(&text) +} + +/// Read an IP + latency CSV (the latency stage output) back into pairs. +fn read_latency_csv(path: &Path) -> Result> { + let text = std::fs::read_to_string(path) + .with_context(|| format!("read latency CSV {}", path.display()))?; + let mut out = Vec::new(); + for line in text.lines() { + let line = line.trim(); + if line.is_empty() || line.starts_with("IP,") { + continue; + } + let mut parts = line.split(','); + let ip = parts.next().and_then(|s| s.trim().parse::().ok()); + let ms = parts.next().and_then(|s| s.trim().parse::().ok()); + if let (Some(ip), Some(ms)) = (ip, ms) { + out.push((ip, Duration::from_secs_f64(ms / 1000.0))); + } + } + Ok(out) +} + +/// Speed-stage table: IP, latency, speed. +fn print_speed_table(results: &[speed::SpeedResult]) { + eprintln!(); + eprintln!( + "{}", + style(format!(" Top {} nodes", results.len())).bold().underlined() + ); + eprintln!(" {:>3} {:<39} {:>9} {:>10}", "#", "IP", "Latency", "Speed"); + for (i, r) in results.iter().enumerate() { + let ms = r.latency.as_secs_f64() * 1000.0; + let speed = format!("{:.2} Mbps", r.speed_mbps); + let colored = if r.speed_mbps >= 20.0 { + style(speed).green() + } else if r.speed_mbps >= 5.0 { + style(speed).yellow() + } else { + style(speed).red() + }; + eprintln!(" {:>3} {:<39} {:>6.0} ms {}", i + 1, r.ip.to_string(), ms, colored); + } + eprintln!(); +} + +/// Speed CSV: IP, latency, speed, best first. +fn write_speed_csv(results: &[speed::SpeedResult], path: &Path) -> Result<()> { + let mut body = String::from("IP,Latency(ms),Speed(Mbps)\n"); + for r in results { + body.push_str(&format!( + "{},{:.2},{:.2}\n", + r.ip, + r.latency.as_secs_f64() * 1000.0, + r.speed_mbps + )); + } + write_file(path, &body)?; + eprintln!("{} saved CSV -> {}", style("✓").green().bold(), style(path.display()).cyan()); + Ok(()) +} + +/// Read an IP list, accepting plain `ip` lines or CSV rows whose first column +/// is the IP. Blank/header/unparseable lines are skipped. 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()) + .filter_map(|line| line.split(',').next()?.trim().parse::().ok()) .collect()) } diff --git a/src/speed.rs b/src/speed.rs new file mode 100644 index 0000000..199388a --- /dev/null +++ b/src/speed.rs @@ -0,0 +1,158 @@ +//! Stage 3 (`speed`): download through the node via each candidate IP and +//! measure real throughput, then rank by speed bucket then latency. + +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}; + +const TARGET_HOST: &str = "cachefly.cachefly.net"; +const TARGET_PORT: u16 = 443; +const TARGET_PATH: &str = "/50mb.test"; + +/// One IP with its (carried-over) latency and measured download speed. +#[derive(Clone, Debug)] +pub struct SpeedResult { + pub ip: IpAddr, + pub latency: Duration, + pub speed_mbps: f64, +} + +/// Progress event, emitted once per finished measurement. +pub struct Probed { + pub ip: IpAddr, + pub speed_mbps: Option, + pub done: usize, + #[allow(dead_code)] // available to callers that want a denominator. + pub total: usize, +} + +/// 1 Mbps-wide bucket used for ranking (higher bucket wins, ties by latency). +pub fn bucket(speed_mbps: f64) -> i64 { + speed_mbps.floor() as i64 +} + +/// Speed-test every input IP, carrying its latency through, and return +/// successes ranked best-first (speed bucket desc, then latency asc). +pub async fn run( + node: &VlessNode, + inputs: &[(IpAddr, Duration)], + concurrency: usize, + timeout: Duration, + limit_bytes: u64, + mut on_progress: impl FnMut(Probed), +) -> Vec { + let total = inputs.len(); + let mut done = 0usize; + let mut results: Vec = Vec::new(); + + let mut stream = stream::iter(inputs.iter().copied().map(|(ip, latency)| async move { + (ip, latency, measure_download(node, ip, timeout, limit_bytes).await) + })) + .buffer_unordered(concurrency.max(1)); + + while let Some((ip, latency, outcome)) = stream.next().await { + done += 1; + let speed = outcome.ok(); + if let Some(mbps) = speed { + results.push(SpeedResult { ip, latency, speed_mbps: mbps }); + } + on_progress(Probed { ip, speed_mbps: speed, done, total }); + } + + results.sort_by(|a, b| { + bucket(b.speed_mbps) + .cmp(&bucket(a.speed_mbps)) + .then(a.latency.cmp(&b.latency)) + }); + results +} + +/// Download up to `limit_bytes` (or until `timeout`) through the tunnel and +/// return throughput in Mbps. Whatever was transferred before the deadline +/// still counts, matching the Python tool's hard time-cap behaviour. +pub async fn measure_download( + node: &VlessNode, + ip: IpAddr, + timeout: Duration, + limit_bytes: u64, +) -> Result { + let tunnel = tokio::time::timeout(timeout, vless::connect(node, ip, TARGET_HOST, TARGET_PORT)) + .await + .map_err(|_| anyhow!("connect timeout"))??; + let sni = ServerName::try_from(TARGET_HOST.to_string())?; + let mut stream = tokio::time::timeout(timeout, vless::tls_connector().connect(sni, tunnel)) + .await + .map_err(|_| anyhow!("tls timeout"))??; + + 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 deadline = Instant::now() + timeout; + let mut buf = vec![0u8; 65536]; + let mut header: Vec = Vec::new(); + let mut header_done = false; + let mut body_bytes: u64 = 0; + let started = Instant::now(); + + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + break; + } + let n = match tokio::time::timeout(remaining, stream.read(&mut buf)).await { + Ok(Ok(0)) => break, // EOF + Ok(Ok(n)) => n, + Ok(Err(e)) => return Err(e.into()), + Err(_) => break, // hit the time cap + }; + + if header_done { + body_bytes += n as u64; + } else { + header.extend_from_slice(&buf[..n]); + if let Some(pos) = find_subsequence(&header, b"\r\n\r\n") { + let status = parse_status(&header).ok_or_else(|| anyhow!("no http status"))?; + if !(200..300).contains(&status) { + return Err(anyhow!("http status {status}")); + } + body_bytes += (header.len() - (pos + 4)) as u64; + header_done = true; + header = Vec::new(); + } else if header.len() > 16384 { + return Err(anyhow!("response header too large")); + } + } + + if body_bytes >= limit_bytes { + break; + } + } + + if body_bytes == 0 { + return Err(anyhow!("no data received")); + } + let elapsed = started.elapsed().as_secs_f64().max(0.001); + Ok(body_bytes as f64 * 8.0 / elapsed / 1_000_000.0) +} + +/// Parse the numeric status code from an HTTP response's first line. +fn parse_status(data: &[u8]) -> Option { + let line_end = find_subsequence(data, b"\r\n")?; + let line = std::str::from_utf8(&data[..line_end]).ok()?; + line.split_whitespace().nth(1)?.parse().ok() +} + +/// Find the first index of `needle` within `haystack`. +fn find_subsequence(haystack: &[u8], needle: &[u8]) -> Option { + haystack.windows(needle.len()).position(|w| w == needle) +} diff --git a/src/vless.rs b/src/vless.rs index be87f75..b3b4c8a 100644 --- a/src/vless.rs +++ b/src/vless.rs @@ -95,6 +95,25 @@ impl VlessNode { remark: url.fragment().unwrap_or("").to_string(), }) } + + /// Build an importable `vless://` URL that connects via `ip` (keeping the + /// original host as SNI/WS-Host) and is tagged with `alias` as its name. + pub fn to_url(&self, ip: IpAddr, alias: &str) -> String { + let query = url::form_urlencoded::Serializer::new(String::new()) + .append_pair("encryption", &self.encryption) + .append_pair("security", "tls") + .append_pair("insecure", "0") + .append_pair("allowInsecure", "0") + .append_pair("type", &self.network) + .append_pair("host", &self.host) + .append_pair("path", &self.path) + .finish(); + let addr = match ip { + IpAddr::V4(v4) => v4.to_string(), + IpAddr::V6(v6) => format!("[{v6}]"), + }; + format!("vless://{}@{}:{}?{}#{}", self.uuid, addr, self.port, query, alias) + } } /// The raw byte stream to the target host, established through the node.