eggress_cli/
lib.rs

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

eggress_cli/
lib.rs
