eggress_cli/
lib.rs

1use std::time::{Duration, Instant};
2
3/// Apply the optional Linux pproxy daemon transition after compatibility
4/// parsing and configuration validation. Re-exec keeps the transition safe
5/// under the workspace's `unsafe_code = "deny"` policy and leaves signal,
6/// listener, and system-proxy rollback ownership with the child process.
7#[cfg(feature = "pproxy-daemon")]
8pub fn maybe_daemonize(requested: bool) -> Result<(), String> {
9    const CHILD_MARKER: &str = "EGGRESS_PPROXY_DAEMON_CHILD";
10    if !requested || std::env::var_os(CHILD_MARKER).is_some() {
11        return Ok(());
12    }
13    #[cfg(target_os = "linux")]
14    {
15        let executable = std::env::current_exe()
16            .map_err(|error| format!("cannot resolve executable for --daemon: {error}"))?;
17        std::process::Command::new(executable)
18            .args(std::env::args_os().skip(1))
19            .env(CHILD_MARKER, "1")
20            .current_dir("/")
21            .stdin(std::process::Stdio::null())
22            .stdout(std::process::Stdio::null())
23            .stderr(std::process::Stdio::null())
24            .spawn()
25            .map_err(|error| format!("cannot start --daemon child: {error}"))?;
26        std::process::exit(0);
27    }
28    #[cfg(not(target_os = "linux"))]
29    {
30        Err("--daemon compatibility is only available on Linux".to_string())
31    }
32}
33
34use eggress_core::chain::{ChainExecutor, HopHandler};
35use eggress_core::{BoxStream, TargetAddr, TargetHost};
36
37#[derive(serde::Serialize)]
38pub struct UpstreamTestResult {
39    pub id: String,
40    pub host: String,
41    pub port: u16,
42    pub target: String,
43    pub mode: String,
44    pub reachable: bool,
45    pub latency_ms: Option<u64>,
46    pub error: Option<String>,
47    pub failure: Option<String>,
48    pub failed_hop: Option<usize>,
49}
50
51/// Parse the URL-shaped `pproxy --test` value into the target address used by
52/// the shared native upstream tester. The regular `eggress upstream test`
53/// command continues to accept its existing `host:port` form.
54pub fn parse_pproxy_test_target(value: &str) -> Result<TargetAddr, String> {
55    if let Ok(target) = value.parse::<TargetAddr>() {
56        return Ok(target);
57    }
58
59    let uri: http::Uri = value
60        .parse()
61        .map_err(|e| format!("invalid test URL '{value}': {e}"))?;
62    let scheme = uri
63        .scheme_str()
64        .ok_or_else(|| format!("invalid test URL '{value}': missing scheme"))?;
65    if !matches!(scheme, "http" | "https") {
66        return Err(format!(
67            "invalid test URL '{value}': unsupported scheme '{scheme}'"
68        ));
69    }
70    let authority = uri
71        .authority()
72        .ok_or_else(|| format!("invalid test URL '{value}': missing host"))?;
73    let host = authority.host();
74    if host.is_empty() {
75        return Err(format!("invalid test URL '{value}': missing host"));
76    }
77    let port = authority
78        .port_u16()
79        .unwrap_or_else(|| if scheme == "https" { 443 } else { 80 });
80    let host = if host.contains(':') {
81        TargetHost::Ip(
82            host.parse()
83                .map_err(|e| format!("invalid test URL '{value}': invalid IPv6 host: {e}"))?,
84        )
85    } else if let Ok(ip) = host.parse() {
86        TargetHost::Ip(ip)
87    } else {
88        TargetHost::Domain(host.to_string())
89    };
90    Ok(TargetAddr { host, port })
91}
92
93/// Run upstream tests against a compiled config and return the exit code.
94///
95/// This is the shared implementation used by both `eggress upstream test`
96/// and `pproxy --test`. It accepts typed inputs, not CLI argv.
97pub fn run_upstream_test(
98    rt: &eggress_config::compile::RuntimeConfig,
99    target: Option<&str>,
100    timeout: Duration,
101    json_output: bool,
102) -> i32 {
103    run_upstream_test_with_mode(rt, target, "proxy", timeout, json_output)
104}
105
106/// Run upstream tests with an explicit mode ("proxy" or "tcp").
107pub fn run_upstream_test_with_mode(
108    rt: &eggress_config::compile::RuntimeConfig,
109    target: Option<&str>,
110    mode: &str,
111    timeout: Duration,
112    json_output: bool,
113) -> i32 {
114    let target = match target {
115        Some(t) => match t.parse::<TargetAddr>() {
116            Ok(addr) => addr,
117            Err(e) => {
118                eprintln!("invalid target: {e}");
119                return 2;
120            }
121        },
122        None => TargetAddr {
123            host: TargetHost::Domain("example.com".to_string()),
124            port: 443,
125        },
126    };
127
128    let target_string = target.to_string();
129    let is_proxy_mode = mode == "proxy";
130    let mut results = Vec::new();
131
132    for upstream in &rt.upstreams {
133        let chain = &upstream.chain;
134        let first_hop = &chain.hops[0];
135        let host = &first_hop.endpoint.host;
136        let port = first_hop.endpoint.port;
137
138        let result = if is_proxy_mode {
139            let hops = chain.hops.clone();
140            let target_for_closure = target.clone();
141            let (reachable, latency_ms, error) = run_async_test(move || {
142                let target = target_for_closure.clone();
143                let hops = hops.clone();
144                Box::pin(async move {
145                    let executor = build_test_chain_executor();
146                    test_upstream_proxy(&executor, &hops, &target, timeout).await
147                })
148            });
149            UpstreamTestResult {
150                id: upstream.id.clone(),
151                host: host.clone(),
152                port,
153                target: target_string.clone(),
154                mode: "proxy".to_string(),
155                reachable,
156                latency_ms,
157                error,
158                failure: None,
159                failed_hop: None,
160            }
161        } else {
162            let host_owned = host.clone();
163            let target_result = run_async_test(move || {
164                let host = host_owned.clone();
165                Box::pin(async move { test_upstream_tcp(&host, port, timeout).await })
166            });
167            UpstreamTestResult {
168                id: upstream.id.clone(),
169                host: host.clone(),
170                port,
171                target: target_string.clone(),
172                ..target_result
173            }
174        };
175        results.push(result);
176    }
177
178    if results.is_empty() {
179        eprintln!("no upstreams found matching criteria");
180        return 3;
181    }
182
183    if json_output {
184        match serde_json::to_string_pretty(&results) {
185            Ok(json) => println!("{json}"),
186            Err(e) => {
187                eprintln!("failed to serialize results: {e}");
188                return 1;
189            }
190        }
191    } else {
192        for result in &results {
193            print_upstream_test_result(result);
194        }
195    }
196
197    if results.iter().any(|r| r.reachable) {
198        0
199    } else {
200        1
201    }
202}
203
204pub fn print_upstream_test_result(result: &UpstreamTestResult) {
205    let status = if result.reachable {
206        "reachable"
207    } else {
208        "unreachable"
209    };
210    let latency = result
211        .latency_ms
212        .map(|ms| format!("{}ms", ms))
213        .unwrap_or_else(|| "n/a".to_string());
214    let error = result
215        .error
216        .as_deref()
217        .map(|e| format!(" ({e})"))
218        .unwrap_or_default();
219
220    println!(
221        "{} {}:{} [{}] latency={}{}",
222        result.id, result.host, result.port, status, latency, error
223    );
224}
225
226pub fn run_async_test<F, T>(make_future: F) -> T
227where
228    F: FnOnce() -> std::pin::Pin<Box<dyn std::future::Future<Output = T> + Send>> + Send + 'static,
229    T: Send + 'static,
230{
231    if tokio::runtime::Handle::try_current().is_ok() {
232        std::thread::Builder::new()
233            .name("eggress-cli-test".to_string())
234            .spawn(move || -> T {
235                let rt = tokio::runtime::Builder::new_multi_thread()
236                    .enable_all()
237                    .build()
238                    .expect("failed to build tokio runtime for cli test");
239                rt.block_on(make_future())
240            })
241            .expect("failed to spawn cli test thread")
242            .join()
243            .expect("cli test thread panicked")
244    } else {
245        let rt = tokio::runtime::Builder::new_current_thread()
246            .enable_all()
247            .build()
248            .expect("failed to build tokio runtime for cli test");
249        rt.block_on(make_future())
250    }
251}
252
253async fn test_upstream_proxy(
254    executor: &ChainExecutor,
255    chain: &[eggress_uri::ProxyHopSpec],
256    target: &TargetAddr,
257    timeout: Duration,
258) -> (bool, Option<u64>, Option<String>) {
259    let start = Instant::now();
260
261    match tokio::time::timeout(timeout, executor.execute(chain, target)).await {
262        Ok(Ok(_stream)) => {
263            let elapsed = start.elapsed().as_millis() as u64;
264            (true, Some(elapsed), None)
265        }
266        Ok(Err(e)) => (false, None, Some(e.to_string())),
267        Err(_) => (false, None, Some("connection timed out".to_string())),
268    }
269}
270
271struct HttpHopHandler;
272
273impl HopHandler for HttpHopHandler {
274    fn protocol(&self) -> eggress_uri::ProtocolSpec {
275        eggress_uri::ProtocolSpec::Http
276    }
277
278    fn handshake<'a>(
279        &'a self,
280        stream: BoxStream,
281        target: &'a TargetAddr,
282        hop: &'a eggress_uri::ProxyHopSpec,
283        _hop_index: usize,
284    ) -> std::pin::Pin<
285        Box<
286            dyn std::future::Future<
287                    Output = Result<BoxStream, Box<dyn std::error::Error + Send + Sync>>,
288                > + Send
289                + 'a,
290        >,
291    > {
292        let auth = hop
293            .credentials
294            .as_ref()
295            .map(|c| (c.username.as_str(), c.password.as_str()));
296        Box::pin(async move {
297            eggress_protocol_http::http_connect(stream, target, auth, &Default::default())
298                .await
299                .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)
300        })
301    }
302}
303
304struct Socks5HopHandler;
305
306impl HopHandler for Socks5HopHandler {
307    fn protocol(&self) -> eggress_uri::ProtocolSpec {
308        eggress_uri::ProtocolSpec::Socks5
309    }
310
311    fn handshake<'a>(
312        &'a self,
313        stream: BoxStream,
314        target: &'a TargetAddr,
315        hop: &'a eggress_uri::ProxyHopSpec,
316        _hop_index: usize,
317    ) -> std::pin::Pin<
318        Box<
319            dyn std::future::Future<
320                    Output = Result<BoxStream, Box<dyn std::error::Error + Send + Sync>>,
321                > + Send
322                + 'a,
323        >,
324    > {
325        let socks_addr = target_to_socks_addr(target);
326        let auth = hop
327            .credentials
328            .as_ref()
329            .map(|c| (c.username.as_str(), c.password.as_str()));
330        Box::pin(async move {
331            eggress_protocol_socks::socks5::client::socks5_connect(stream, &socks_addr, auth)
332                .await
333                .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)
334        })
335    }
336}
337
338struct Socks4HopHandler;
339
340impl HopHandler for Socks4HopHandler {
341    fn protocol(&self) -> eggress_uri::ProtocolSpec {
342        eggress_uri::ProtocolSpec::Socks4
343    }
344
345    fn handshake<'a>(
346        &'a self,
347        stream: BoxStream,
348        target: &'a TargetAddr,
349        hop: &'a eggress_uri::ProxyHopSpec,
350        _hop_index: usize,
351    ) -> std::pin::Pin<
352        Box<
353            dyn std::future::Future<
354                    Output = Result<BoxStream, Box<dyn std::error::Error + Send + Sync>>,
355                > + Send
356                + 'a,
357        >,
358    > {
359        let user_id = hop.credentials.as_ref().map(|c| c.username.as_str());
360        Box::pin(async move {
361            eggress_protocol_socks::socks4_connect(stream, target, user_id)
362                .await
363                .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)
364        })
365    }
366}
367
368pub fn build_test_chain_executor() -> ChainExecutor {
369    let handlers: Vec<Box<dyn HopHandler>> = vec![
370        Box::new(HttpHopHandler),
371        Box::new(Socks5HopHandler),
372        Box::new(Socks4HopHandler),
373    ];
374    ChainExecutor::new(handlers)
375}
376
377fn target_to_socks_addr(target: &TargetAddr) -> eggress_protocol_socks::socks5::server::SocksAddr {
378    use eggress_protocol_socks::socks5::server::SocksAddr;
379    match &target.host {
380        TargetHost::Ip(std::net::IpAddr::V4(ip)) => SocksAddr::IPv4(ip.octets(), target.port),
381        TargetHost::Ip(std::net::IpAddr::V6(ip)) => SocksAddr::IPv6(ip.octets(), target.port),
382        TargetHost::Domain(d) => SocksAddr::Domain(d.clone(), target.port),
383    }
384}
385
386pub async fn test_upstream_tcp(host: &str, port: u16, timeout: Duration) -> UpstreamTestResult {
387    let addr = format!("{}:{}", host, port);
388    let start = Instant::now();
389
390    let result = tokio::time::timeout(timeout, tokio::net::TcpStream::connect(&addr)).await;
391
392    let elapsed = start.elapsed().as_millis() as u64;
393
394    match result {
395        Ok(Ok(_stream)) => UpstreamTestResult {
396            id: String::new(),
397            host: host.to_string(),
398            port,
399            target: String::new(),
400            mode: "tcp".to_string(),
401            reachable: true,
402            latency_ms: Some(elapsed),
403            error: None,
404            failure: None,
405            failed_hop: None,
406        },
407        Ok(Err(e)) => UpstreamTestResult {
408            id: String::new(),
409            host: host.to_string(),
410            port,
411            target: String::new(),
412            mode: "tcp".to_string(),
413            reachable: false,
414            latency_ms: None,
415            error: Some(e.to_string()),
416            failure: None,
417            failed_hop: None,
418        },
419        Err(_) => UpstreamTestResult {
420            id: String::new(),
421            host: host.to_string(),
422            port,
423            target: String::new(),
424            mode: "tcp".to_string(),
425            reachable: false,
426            latency_ms: None,
427            error: Some("connection timed out".to_string()),
428            failure: None,
429            failed_hop: None,
430        },
431    }
432}
eggress_cli/lib.rs

eggress_cli/
lib.rs
