From 9a64f9ad0879b382ac23c03b84451f17020875a7 Mon Sep 17 00:00:00 2001 From: Robin B Date: Fri, 18 Sep 2026 21:54:49 +0200 Subject: [PATCH 1/2] fix: replace CLI panics with clean errors on bad IP args and closed stdout - Validate --ipv4/--ipv6 values with clap value parsers so invalid addresses produce a usage error with exit code 2 instead of a panic (previously exit 101), and collapse the three duplicate client construction branches in main.rs into one. - Route all stdout writes through new tolerant helpers (src/stdout.rs): a broken pipe (e.g. the consumer of 'cfspeedtest -o json | jq' exiting early) is now tolerated silently instead of panicking with exit 101; unexpected stdout write errors are reported once on stderr. Completion scripts are buffered in memory because clap_complete panics on write errors. - Add regression tests for both behaviors. --- src/lib.rs | 86 +++++++++++++++++++++++++++++++++++++++++++-- src/main.rs | 42 +++++++++++----------- src/measurements.rs | 68 ++++++++++++++++++++++++----------- src/progress.rs | 8 ++--- src/speedtest.rs | 40 +++++++++------------ src/stdout.rs | 60 +++++++++++++++++++++++++++++++ tests/p1_cli.rs | 77 ++++++++++++++++++++++++++++++++++++++++ 7 files changed, 308 insertions(+), 73 deletions(-) create mode 100644 src/stdout.rs diff --git a/src/lib.rs b/src/lib.rs index 01d7e88..a577b05 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,8 +3,10 @@ pub mod measurements; pub mod progress; pub mod run; pub mod speedtest; +pub mod stdout; use std::fmt; use std::fmt::Display; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; use clap::Parser; use clap_complete::Shell; @@ -63,11 +65,11 @@ pub struct SpeedTestCLIOptions { pub verbose: bool, /// Force IPv4 with provided source IPv4 address or the default IPv4 address bound to the main interface - #[clap(long, value_name = "IPv4", num_args = 0..=1, default_missing_value = "0.0.0.0", conflicts_with = "ipv6")] + #[clap(long, value_name = "IPv4", num_args = 0..=1, default_missing_value = "0.0.0.0", value_parser = parse_ipv4_address, conflicts_with = "ipv6")] pub ipv4: Option, /// Force IPv6 with provided source IPv6 address or the default IPv6 address bound to the main interface - #[clap(long, value_name = "IPv6", num_args = 0..=1, default_missing_value = "::", conflicts_with = "ipv4")] + #[clap(long, value_name = "IPv6", num_args = 0..=1, default_missing_value = "::", value_parser = parse_ipv6_address, conflicts_with = "ipv4")] pub ipv6: Option, /// Disables dynamically skipping tests with larger payload sizes if the tests for the previous payload @@ -104,10 +106,45 @@ fn parse_payload_size(input_string: &str) -> Result { PayloadSize::from(input_string.to_string()) } +fn parse_ipv4_address(input_string: &str) -> Result { + match input_string.parse::() { + Ok(_) => Ok(input_string.to_string()), + Err(_) => Err(format!("invalid IPv4 address: '{input_string}'")), + } +} + +fn parse_ipv6_address(input_string: &str) -> Result { + match input_string.parse::() { + Ok(_) => Ok(input_string.to_string()), + Err(_) => Err(format!("invalid IPv6 address: '{input_string}'")), + } +} + fn parse_output_format(input_string: &str) -> Result { OutputFormat::from(input_string.to_string()) } +/// Parses the bound address from the CLI options. Returns an error if the +/// provided address is not a valid IP address. +pub fn parse_bound_address( + ipv4: &Option, + ipv6: &Option, +) -> Result, String> { + if let Some(address) = ipv4 { + address + .parse::() + .map(Some) + .map_err(|_| format!("invalid IPv4 address: '{address}'")) + } else if let Some(address) = ipv6 { + address + .parse::() + .map(Some) + .map_err(|_| format!("invalid IPv6 address: '{address}'")) + } else { + Ok(None) + } +} + #[cfg(test)] mod tests { use super::*; @@ -169,6 +206,51 @@ mod tests { assert_eq!(format!("{}", OutputFormat::None), "None"); } + #[test] + fn test_parse_ipv4_address_accepts_valid_addresses() { + assert_eq!(parse_ipv4_address("0.0.0.0"), Ok("0.0.0.0".to_string())); + assert_eq!( + parse_ipv4_address("192.168.1.5"), + Ok("192.168.1.5".to_string()) + ); + } + + #[test] + fn test_parse_ipv4_address_rejects_invalid_addresses() { + assert!(parse_ipv4_address("not-an-ip").is_err()); + assert!(parse_ipv4_address("999.1.1.1").is_err()); + // IPv6 addresses belong to --ipv6, not --ipv4. + assert!(parse_ipv4_address("::1").is_err()); + } + + #[test] + fn test_parse_ipv6_address_accepts_valid_addresses() { + assert_eq!(parse_ipv6_address("::"), Ok("::".to_string())); + assert_eq!(parse_ipv6_address("fe80::1"), Ok("fe80::1".to_string())); + } + + #[test] + fn test_parse_ipv6_address_rejects_invalid_addresses() { + assert!(parse_ipv6_address("not-an-ip").is_err()); + // IPv4 addresses belong to --ipv4, not --ipv6. + assert!(parse_ipv6_address("192.168.1.5").is_err()); + } + + #[test] + fn test_parse_bound_address() { + assert_eq!(parse_bound_address(&None, &None), Ok(None)); + assert_eq!( + parse_bound_address(&Some("127.0.0.1".to_string()), &None), + Ok(Some("127.0.0.1".parse().unwrap())) + ); + assert_eq!( + parse_bound_address(&None, &Some("fe80::1".to_string())), + Ok(Some("fe80::1".parse().unwrap())) + ); + assert!(parse_bound_address(&Some("not-an-ip".to_string()), &None).is_err()); + assert!(parse_bound_address(&None, &Some("not-an-ip".to_string())).is_err()); + } + #[test] fn test_cli_options_should_download() { let mut options = SpeedTestCLIOptions { diff --git a/src/main.rs b/src/main.rs index 9736686..0c63067 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,10 +1,9 @@ use cfspeedtest::speedtest; +use cfspeedtest::stdout; use cfspeedtest::OutputFormat; use cfspeedtest::SpeedTestCLIOptions; use clap::{CommandFactory, Parser}; use clap_complete::generate; -use std::io; -use std::net::IpAddr; use cfspeedtest::run::RunConfig; use speedtest::speed_test_with_config; @@ -29,7 +28,11 @@ struct CliOptions { } fn print_completions(gen: G, cmd: &mut clap::Command) { - generate(gen, cmd, cmd.get_name().to_string(), &mut io::stdout()); + // Buffer the script in memory: clap_complete panics on write errors, and + // piping the output into e.g. `head` must not abort the process. + let mut buffer = Vec::new(); + generate(gen, cmd, cmd.get_name().to_string(), &mut buffer); + stdout::print(&String::from_utf8_lossy(&buffer)); } fn main() -> ExitCode { @@ -45,27 +48,22 @@ fn main() -> ExitCode { } if options.output_format == OutputFormat::StdOut { - println!("Starting Cloudflare speed test"); + stdout::print_line("Starting Cloudflare speed test"); } - let client; - if let Some(ref ip) = options.ipv4 { - client = reqwest::blocking::Client::builder() - .local_address(ip.parse::().expect("Invalid IPv4 address")) - .timeout(std::time::Duration::from_secs(30)) - .cookie_store(true) - .build(); - } else if let Some(ref ip) = options.ipv6 { - client = reqwest::blocking::Client::builder() - .local_address(ip.parse::().expect("Invalid IPv6 address")) - .timeout(std::time::Duration::from_secs(30)) - .cookie_store(true) - .build(); - } else { - client = reqwest::blocking::Client::builder() - .timeout(std::time::Duration::from_secs(30)) - .cookie_store(true) - .build(); + let local_address = match cfspeedtest::parse_bound_address(&options.ipv4, &options.ipv6) { + Ok(address) => address, + Err(message) => { + eprintln!("Error: {message}"); + return ExitCode::FAILURE; + } + }; + let mut client_builder = reqwest::blocking::Client::builder() + .timeout(Duration::from_secs(30)) + .cookie_store(true); + if let Some(address) = local_address { + client_builder = client_builder.local_address(address); } + let client = client_builder.build(); let config = RunConfig { base_url: cli.server.as_str().trim_end_matches('/').to_string(), max_duration: Duration::from_secs(cli.max_duration), diff --git a/src/measurements.rs b/src/measurements.rs index 409ccef..12e32a4 100644 --- a/src/measurements.rs +++ b/src/measurements.rs @@ -2,6 +2,7 @@ use crate::boxplot; use crate::run::SpeedTestReport; use crate::speedtest::Metadata; use crate::speedtest::TestType; +use crate::stdout; use crate::OutputFormat; use indexmap::IndexSet; use serde::Serialize; @@ -69,11 +70,13 @@ pub(crate) fn log_measurements( let measurements = &report.measurements; let payload_attempt_stats = &report.payload_attempt_stats; if output_format == OutputFormat::StdOut { - println!("\nSummary Statistics"); + stdout::print_line("\nSummary Statistics"); if verbose { - println!("Type Payload | min/max/avg in mbit/s | attempts/success/skipped"); + stdout::print_line( + "Type Payload | min/max/avg in mbit/s | attempts/success/skipped", + ); } else { - println!("Type Payload | min/max/avg in mbit/s"); + stdout::print_line("Type Payload | min/max/avg in mbit/s"); } } let mut stat_measurements: Vec = Vec::new(); @@ -99,25 +102,50 @@ pub(crate) fn log_measurements( OutputFormat::Csv => { let mut wtr = csv::Writer::from_writer(io::stdout()); for measurement in &stat_measurements { - wtr.serialize(measurement).unwrap(); + if let Err(error) = wtr.serialize(measurement) { + handle_csv_write_error(error); + } + } + if let Err(error) = wtr.flush() { + stdout::handle_write_error(&error); } - wtr.flush().unwrap(); } OutputFormat::Json => { let output = compose_report_json(&stat_measurements, report); - serde_json::to_writer(io::stdout(), &output).unwrap(); - println!(); + if let Err(error) = serde_json::to_writer(io::stdout(), &output) { + handle_json_write_error(error); + } + stdout::print_line(""); } OutputFormat::JsonPretty => { let output = compose_report_json(&stat_measurements, report); - serde_json::to_writer_pretty(io::stdout(), &output).unwrap(); - println!(); + if let Err(error) = serde_json::to_writer_pretty(io::stdout(), &output) { + handle_json_write_error(error); + } + stdout::print_line(""); } OutputFormat::StdOut => {} OutputFormat::None => {} } } +/// Reports a CSV serialization/write failure without panicking: broken pipes +/// are tolerated (the stdout consumer is gone), other errors are reported. +fn handle_csv_write_error(error: csv::Error) { + match error.into_kind() { + csv::ErrorKind::Io(io_error) => stdout::handle_write_error(&io_error), + other => eprintln!("Failed to write CSV output: {other:?}"), + } +} + +/// Reports a JSON serialization/write failure without panicking. +fn handle_json_write_error(error: serde_json::Error) { + stdout::handle_write_kind( + error.io_error_kind().unwrap_or(io::ErrorKind::Other), + &format!("Failed to write JSON output: {error}"), + ); +} + fn compose_report_json( stat_measurements: &[StatMeasurement], report: &SpeedTestReport, @@ -222,22 +250,22 @@ fn log_measurements_by_test_type( }); if output_format == OutputFormat::StdOut { if verbose { - println!( + stdout::print_line(&format!( "{fmt_test_type:<9} {formatted_payload:<7}| min {min:<7.2} max {max:<7.2} avg {avg:<7.2} | {attempts:>3}/{successes:>3}/{skipped:>3}" - ); + )); } else { - println!( + stdout::print_line(&format!( "{fmt_test_type:<9} {formatted_payload:<7}| min {min:<7.2} max {max:<7.2} avg {avg:<7.2}" - ); + )); } if successes < target_successes { - println!( + stdout::print_line(&format!( " insufficient samples: collected {successes}/{target_successes} successful runs" - ); + )); } if verbose { let plot = boxplot::render_plot(min, q1, median, q3, max); - println!("{plot}\n"); + stdout::print_line(&format!("{plot}\n")); } } } else { @@ -257,13 +285,13 @@ fn log_measurements_by_test_type( }); if output_format == OutputFormat::StdOut { if verbose { - println!( + stdout::print_line(&format!( "{fmt_test_type:<9} {formatted_payload:<7}| min N/A max N/A avg N/A | {attempts:>3}/{successes:>3}/{skipped:>3} (insufficient samples)" - ); + )); } else { - println!( + stdout::print_line(&format!( "{fmt_test_type:<9} {formatted_payload:<7}| min N/A max N/A avg N/A (insufficient samples)" - ); + )); } } } diff --git a/src/progress.rs b/src/progress.rs index 9c6d150..4dacc70 100644 --- a/src/progress.rs +++ b/src/progress.rs @@ -1,15 +1,13 @@ -use std::io::stdout; -use std::io::Write; +use crate::stdout; pub fn print_progress(name: &str, curr: u32, max: u32) { const BAR_LEN: u32 = 30; let progress_line = ((curr as f32 / max as f32) * BAR_LEN as f32) as u32; let remaining_line = BAR_LEN - progress_line; - print!( + stdout::print(&format!( "\r{:<15} [{}{}]", name, (0..progress_line).map(|_| "=").collect::(), (0..remaining_line).map(|_| "-").collect::(), - ); - stdout().flush().expect("error printing progress bar"); + )); } diff --git a/src/speedtest.rs b/src/speedtest.rs index e48febb..cfa3180 100644 --- a/src/speedtest.rs +++ b/src/speedtest.rs @@ -7,6 +7,7 @@ use crate::run::{ interruptible_sleep, LatencyReport, LatencyStatus, MeasurementError, RunConfig, RunControl, RunStatus, SpeedTestReport, }; +use crate::stdout; use crate::OutputFormat; use crate::SpeedTestCLIOptions; use jiff::Zoned; @@ -16,7 +17,6 @@ use reqwest::{blocking::Client, header::RETRY_AFTER, StatusCode}; use serde::Serialize; use std::{ fmt::Display, - io::Write, sync::{ atomic::{AtomicBool, Ordering}, LazyLock, @@ -136,7 +136,7 @@ pub fn speed_test_with_config( }; if options.output_format == OutputFormat::StdOut { if let Some(metadata) = &metadata { - println!("{metadata}"); + stdout::print_line(&metadata.to_string()); } } let latency = run_latency_with_control( @@ -265,15 +265,15 @@ fn run_latency_with_control( report.finish(); if output_format == OutputFormat::StdOut && nr_latency_tests > 0 { if let Some(avg) = report.avg_latency_ms { - println!( + stdout::print_line(&format!( "\nAvg GET request latency {avg:.2} ms ({}/{} valid samples)\n", report.successes, report.target_samples - ); + )); } else { - println!( + stdout::print_line(&format!( "\nAvg GET request latency N/A (0/{} valid samples)\n", report.target_samples - ); + )); } } report @@ -428,7 +428,7 @@ pub fn run_tests( nr_tests, nr_tests, ); - println!(); + stdout::print_line(""); } if !disable_dynamic_max_payload_size && start.elapsed() > TIME_THRESHOLD { log::info!("Exceeded threshold"); @@ -625,7 +625,7 @@ fn run_tests_with_control( if options.output_format == OutputFormat::StdOut { print_progress(&label, successes, options.nr_tests); - println!(); + stdout::print_line(""); } payload_attempt_stats.push(PayloadAttemptStats { @@ -931,45 +931,41 @@ fn compute_retry_delay(retry_count: u32, retry_after: Option) -> Durat } fn print_current_speed(mbits: f64, duration: Duration, payload_size_bytes: usize) { - print!( + stdout::print(&format!( " {:>6.2} mbit/s | {:>5} in {:>4}ms ", mbits, format_bytes(payload_size_bytes), duration.as_millis(), - ); - flush_stdout(); + )); } fn print_skipped_sample(duration: Duration, status_code: StatusCode, payload_size_bytes: usize) { - print!( + stdout::print(&format!( " {:>6} mbit/s | {:>5} in {:>4}ms -> status: {} ", "N/A", format_bytes(payload_size_bytes), duration.as_millis(), status_code - ); - flush_stdout(); + )); } fn print_retry_notice(delay: Duration, attempt: u32, max_attempts: u32) { let delay_display = format_retry_delay(delay); let eta_display = format_retry_eta(delay); - print!( + stdout::print(&format!( " retrying in {}{} ({}/{}) ", delay_display, eta_display, attempt, max_attempts - ); - flush_stdout(); + )); } fn print_transport_failure(duration: Duration, payload_size_bytes: usize, error: &reqwest::Error) { - print!( + stdout::print(&format!( " {:>6} mbit/s | {:>5} in {:>4}ms -> error: {} ", "N/A", format_bytes(payload_size_bytes), duration.as_millis(), error - ); - flush_stdout(); + )); } fn format_retry_delay(delay: Duration) -> String { @@ -996,10 +992,6 @@ fn format_retry_eta(delay: Duration) -> String { format!(" (until {})", eta.strftime("%H:%M:%S %Z")) } -fn flush_stdout() { - let _ = std::io::stdout().flush(); -} - pub fn fetch_metadata(client: &Client) -> Result { fetch_metadata_request(client, BASE_URL, None) } diff --git a/src/stdout.rs b/src/stdout.rs new file mode 100644 index 0000000..b76cfdd --- /dev/null +++ b/src/stdout.rs @@ -0,0 +1,60 @@ +//! Helpers for writing terminal output to stdout without panicking. +//! +//! This CLI's reports are routinely piped into other tools, e.g. +//! `cfspeedtest -o json | jq`. When such a consumer exits before the run +//! finishes, stdout writes fail with [`io::ErrorKind::BrokenPipe`]. The +//! `print!`/`println!` macros panic on any write error, so all stdout output +//! goes through these helpers instead: a broken pipe is tolerated silently +//! because the consumer is simply gone, while unexpected write errors are +//! reported once on stderr (the diagnostic channel) without aborting the run. + +use std::io::{self, Write}; +use std::sync::atomic::{AtomicBool, Ordering}; + +static REPORTED_WRITE_ERROR: AtomicBool = AtomicBool::new(false); + +/// Writes `text` to stdout (without trailing newline) and flushes. +pub fn print(text: &str) { + let mut stdout = io::stdout().lock(); + if let Err(error) = stdout + .write_all(text.as_bytes()) + .and_then(|()| stdout.flush()) + { + handle_write_error(&error); + } +} + +/// Writes `text` followed by a newline to stdout and flushes. +pub fn print_line(text: &str) { + print(&format!("{text}\n")); +} + +/// Reports a failed stdout write. Broken pipes are ignored; any other error +/// is reported on stderr, at most once per process. +pub fn handle_write_error(error: &io::Error) { + handle_write_kind(error.kind(), &format!("Failed to write to stdout: {error}")); +} + +/// Reports a failed stdout write by error kind. Broken pipes are ignored; +/// any other error is reported on stderr, at most once per process. +pub fn handle_write_kind(kind: io::ErrorKind, message: &str) { + if kind == io::ErrorKind::BrokenPipe { + // The consumer of our stdout went away; keep going without output. + return; + } + if !REPORTED_WRITE_ERROR.swap(true, Ordering::Relaxed) { + eprintln!("{message}"); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn broken_pipe_is_tolerated() { + // The handler must classify a broken pipe as tolerable instead of + // panicking; there is no stdout to assert on. + handle_write_kind(io::ErrorKind::BrokenPipe, "unreachable"); + } +} diff --git a/tests/p1_cli.rs b/tests/p1_cli.rs index f6cd9c8..5c37fb0 100644 --- a/tests/p1_cli.rs +++ b/tests/p1_cli.rs @@ -1,6 +1,7 @@ mod support; use serde_json::Value; +use std::process::Stdio; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -16,6 +17,82 @@ fn json(output: &std::process::Output) -> Value { }) } +#[test] +fn p1_cli_invalid_ip_arguments_are_usage_errors() { + for (flag, value, expected) in [ + ("--ipv4", "not-an-ip", "invalid IPv4 address"), + ("--ipv6", "not-an-ip", "invalid IPv6 address"), + // An IPv6 address is not valid for --ipv4 and vice versa. + ("--ipv4", "::1", "invalid IPv4 address"), + ("--ipv6", "127.0.0.1", "invalid IPv6 address"), + ] { + let output = std::process::Command::new(env!("CARGO_BIN_EXE_cfspeedtest")) + .args([flag, value]) + .env("NO_PROXY", "*") + .env("no_proxy", "*") + .output() + .unwrap(); + assert_eq!( + output.status.code(), + Some(2), + "unexpected exit status for {flag} {value}" + ); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains(expected), + "missing '{expected}' for {flag} {value}: {stderr}" + ); + assert!(!String::from_utf8_lossy(&output.stdout).contains("panic")); + } +} + +#[cfg(unix)] +#[test] +fn p1_cli_closed_stdout_is_not_a_panic() { + let server = Server::new(|path| Response::normal(path)); + + // JSON mode writes the report once at the end of the run. + let mut child = server.command(&["-o", "json"]).spawn().unwrap(); + drop(child.stdout.take()); + let output = child.wait_with_output().unwrap(); + assert_eq!(output.status.code(), Some(0)); + assert!( + output.stderr.is_empty(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + + // Human mode writes progress continuously during the run. + let mut child = server.command(&["-o", "stdout"]).spawn().unwrap(); + drop(child.stdout.take()); + let output = child.wait_with_output().unwrap(); + assert_eq!(output.status.code(), Some(0)); + assert!( + output.stderr.is_empty(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); +} + +#[cfg(unix)] +#[test] +fn p1_cli_completion_output_survives_closed_stdout() { + let mut child = std::process::Command::new(env!("CARGO_BIN_EXE_cfspeedtest")) + .args(["--generate-completion", "bash"]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + drop(child.stdout.take()); + let output = child.wait_with_output().unwrap(); + assert_eq!(output.status.code(), Some(0)); + assert!( + !String::from_utf8_lossy(&output.stderr).contains("panic"), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); +} + #[test] fn p1_cli_all_speed_samples_failed_is_nonzero() { let server = Server::new(|path| { From feb0c922c1cedebd848a586c125498012178356b Mon Sep 17 00:00:00 2001 From: Robin B Date: Fri, 18 Sep 2026 22:04:24 +0200 Subject: [PATCH 2/2] fix redundant closure flagged by clippy --- tests/p1_cli.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/p1_cli.rs b/tests/p1_cli.rs index 5c37fb0..9d8bbf8 100644 --- a/tests/p1_cli.rs +++ b/tests/p1_cli.rs @@ -49,7 +49,7 @@ fn p1_cli_invalid_ip_arguments_are_usage_errors() { #[cfg(unix)] #[test] fn p1_cli_closed_stdout_is_not_a_panic() { - let server = Server::new(|path| Response::normal(path)); + let server = Server::new(Response::normal); // JSON mode writes the report once at the end of the run. let mut child = server.command(&["-o", "json"]).spawn().unwrap();