eggress_server/
lib.rs

1pub mod accept;
2#[cfg(feature = "extended")]
3pub mod advanced;
4mod auth;
5pub mod error;
6pub mod execute;
7pub mod listener;
8pub mod reply;
9
10use std::sync::Arc;
11use std::time::Duration;
12
13pub use accept::AcceptedSession;
14pub use accept::AuthReuseCache;
15pub use error::SessionOpenError;
16pub use execute::{build_chain_executor, FailureCategory, SessionReport};
17
18use eggress_routing::RouteService;
19
20/// Trait for recording session metrics. Implemented by external crates.
21pub trait SessionMetrics: Send + Sync {
22    fn record_session_start(&self);
23    fn record_session(&self, report: &SessionReport);
24    fn record_route_decision(&self, rule: &str, action: &str, outcome: &str);
25    fn record_upstream_open(&self, protocol: &str, outcome: &str);
26    fn record_upstream_failure(&self, protocol: &str, reason: &str);
27    fn record_auth_failure(&self);
28    fn record_platform_capability_check_failure(&self) {}
29    fn record_unix_listener_connection_accepted(&self) {}
30    fn record_reload(&self, _success: bool) {}
31    fn set_config_generation(&self, _generation: u64) {}
32    fn record_udp_association_created(&self) {}
33    fn render_prometheus(&self) -> String {
34        String::new()
35    }
36}
37
38/// No-op implementation of SessionMetrics for builds without operations support.
39pub struct NoopMetrics;
40
41impl SessionMetrics for NoopMetrics {
42    fn record_session_start(&self) {}
43    fn record_session(&self, _report: &SessionReport) {}
44    fn record_route_decision(&self, _rule: &str, _action: &str, _outcome: &str) {}
45    fn record_upstream_open(&self, _protocol: &str, _outcome: &str) {}
46    fn record_upstream_failure(&self, _protocol: &str, _reason: &str) {}
47    fn record_auth_failure(&self) {}
48}
49
50/// Handle returned by UdpService::create_association.
51pub struct UdpAssociationHandle {
52    pub id: eggress_udp::assoc::UdpAssociationId,
53    pub relay_addr: std::net::SocketAddr,
54    pub cancel: tokio_util::sync::CancellationToken,
55}
56
57/// Trait for UDP association services. Implemented by the runtime crate.
58pub trait UdpService: Send + Sync {
59    fn create_association(
60        &self,
61        listener: &str,
62        client_tcp_peer: Option<std::net::SocketAddr>,
63        identity: eggress_core::ClientIdentity,
64        generation: u64,
65    ) -> std::pin::Pin<
66        Box<
67            dyn std::future::Future<
68                    Output = Result<UdpAssociationHandle, eggress_udp::error::UdpError>,
69                > + Send
70                + 'static,
71        >,
72    >;
73    fn is_enabled(&self) -> bool;
74    fn active_count(
75        &self,
76    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = usize> + Send + 'static>>;
77}
78
79/// Context propagated from the listener into routing decisions.
80#[derive(Clone, Default)]
81pub struct ConnectionContext {
82    pub source: Option<std::net::SocketAddr>,
83    pub listener: String,
84    pub generation: u64,
85}
86
87/// Configuration for a single connection.
88#[derive(Clone)]
89pub struct ConnectionConfig {
90    pub routing: Arc<dyn RouteService>,
91    pub context: ConnectionContext,
92    pub handshake_timeout: Duration,
93    pub connect_timeout: Duration,
94    pub protocols: Arc<[eggress_core::ProtocolId]>,
95    pub authentication: accept::InboundAuthentication,
96    pub metrics: Option<Arc<dyn SessionMetrics>>,
97    pub udp: Option<Arc<dyn UdpService>>,
98    /// Optional TLS client config override for upstream connections (e.g., Trojan).
99    /// When `None`, the chain executor builds a config with system root CAs.
100    /// Intended for test-only use (e.g., insecure TLS for self-signed certs).
101    pub tls_client_config: Option<Arc<rustls::ClientConfig>>,
102    pub shadowsocks: Option<accept::InboundShadowsocksConfig>,
103    /// Optional Trojan inbound configuration for password verification on Trojan listeners.
104    pub trojan: Option<accept::InboundTrojanConfig>,
105    pub fixed_target: Option<eggress_core::TargetAddr>,
106    pub local_bind: Option<String>,
107    /// Shared optional SSH session cache for compatibility upstreams.
108    #[cfg(feature = "ssh")]
109    pub ssh_sessions: Option<Arc<eggress_transport_ssh::SshSessionCache>>,
110    /// Optional Shadowsocks-specific metrics for observability.
111    #[cfg(feature = "extended")]
112    pub shadowsocks_metrics: Option<Arc<eggress_protocol_shadowsocks::ShadowsocksMetrics>>,
113    #[cfg(not(feature = "extended"))]
114    pub shadowsocks_metrics: Option<()>,
115}
116
117/// Handle a single inbound connection.
118///
119/// Every non-panicking return from this function goes through exactly one
120/// terminal metrics finalization: after `record_session_start()`, exactly
121/// one `record_session()` call is made before returning.
122pub async fn serve_connection(
123    client: eggress_core::BoxStream,
124    config: ConnectionConfig,
125) -> SessionReport {
126    // Buffer reads so protocol handshakes that consume the head
127    // incrementally do not issue one syscall per byte. Unconsumed
128    // prefetch stays available to later reads on the same stream.
129    let client: eggress_core::BoxStream = Box::new(tokio::io::BufReader::new(client));
130
131    if let Some(metrics) = &config.metrics {
132        metrics.record_session_start();
133    }
134
135    let accepted = tokio::time::timeout(
136        config.handshake_timeout,
137        accept::accept_with_fixed_target_for_peer(
138            client,
139            &config.protocols,
140            &config.authentication,
141            config.shadowsocks.as_ref(),
142            config.shadowsocks_metrics.as_ref(),
143            config.trojan.as_ref(),
144            config.fixed_target.as_ref(),
145            config.context.source.map(|peer| peer.ip()),
146        ),
147    )
148    .await;
149
150    let report = match accepted {
151        Ok(Ok(session)) => execute::execute(session, &config).await,
152        Ok(Err(accept::AcceptError::AuthenticationFailed)) => {
153            if let Some(metrics) = &config.metrics {
154                metrics.record_auth_failure();
155            }
156            SessionReport {
157                protocol: None,
158                target: None,
159                route: "unknown".to_string(),
160                bytes_upstream: 0,
161                bytes_downstream: 0,
162                outcome: execute::SessionOutcome::AuthenticationFailed,
163                failure: Some(execute::FailureCategory::Authentication),
164                rule_id: None,
165                upstream_group: None,
166                upstream_id: None,
167                selection_reason: None,
168            }
169        }
170        Ok(Err(_)) => SessionReport {
171            protocol: None,
172            target: None,
173            route: "unknown".to_string(),
174            bytes_upstream: 0,
175            bytes_downstream: 0,
176            outcome: execute::SessionOutcome::ClientProtocolError,
177            failure: Some(execute::FailureCategory::Protocol),
178            rule_id: None,
179            upstream_group: None,
180            upstream_id: None,
181            selection_reason: None,
182        },
183        Err(_) => SessionReport {
184            protocol: None,
185            target: None,
186            route: "unknown".to_string(),
187            bytes_upstream: 0,
188            bytes_downstream: 0,
189            outcome: execute::SessionOutcome::HandshakeTimedOut,
190            failure: Some(execute::FailureCategory::HandshakeTimeout),
191            rule_id: None,
192            upstream_group: None,
193            upstream_id: None,
194            selection_reason: None,
195        },
196    };
197
198    if let Some(metrics) = &config.metrics {
199        metrics.record_session(&report);
200    }
201
202    #[cfg(feature = "extended")]
203    if let Some(ss_metrics) = &config.shadowsocks_metrics {
204        if report.protocol.as_deref() == Some("shadowsocks") {
205            ss_metrics.record_tcp_session_closed();
206            ss_metrics.record_tcp_flow_close();
207        }
208    }
209
210    tracing::info!(
211        outcome = ?report.outcome,
212        failure = ?report.failure,
213        protocol = ?report.protocol,
214        target = ?report.target,
215        route = %report.route,
216        rule = ?report.rule_id,
217        upstream_group = ?report.upstream_group,
218        upstream = ?report.upstream_id,
219        selection_reason = ?report.selection_reason,
220        bytes_upstream = report.bytes_upstream,
221        bytes_downstream = report.bytes_downstream,
222        "connection completed",
223    );
224
225    report
226}
227
228#[cfg(test)]
229mod tests {
230    use super::*;
231    use eggress_routing::{RouteActionSpec, Router};
232    use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
233
234    fn all_protocols() -> Arc<[eggress_core::ProtocolId]> {
235        Arc::from([
236            eggress_core::ProtocolId::Http,
237            eggress_core::ProtocolId::Socks4,
238            eggress_core::ProtocolId::Socks5,
239            eggress_core::ProtocolId::Http2,
240            eggress_core::ProtocolId::WebSocket,
241            eggress_core::ProtocolId::Raw,
242        ])
243    }
244
245    fn direct_routing() -> Arc<dyn RouteService> {
246        Arc::new(Router::new(vec![], RouteActionSpec::Direct))
247    }
248
249    fn test_config(routing: Arc<dyn RouteService>) -> ConnectionConfig {
250        ConnectionConfig {
251            routing,
252            context: ConnectionContext::default(),
253            handshake_timeout: Duration::from_secs(5),
254            connect_timeout: Duration::from_secs(10),
255            protocols: all_protocols(),
256            authentication: accept::InboundAuthentication::None,
257            metrics: None,
258            udp: None,
259            tls_client_config: None,
260            shadowsocks: None,
261            shadowsocks_metrics: None,
262            trojan: None,
263            fixed_target: None,
264            local_bind: None,
265        }
266    }
267
268    #[tokio::test]
269    async fn test_serve_connection_socks5_direct() {
270        let (echo_addr, echo_jh) = eggress_testkit::start_echo_server().await;
271
272        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
273        let proxy_addr = proxy_listener.local_addr().unwrap();
274
275        let proxy_jh = tokio::spawn(async move {
276            let (stream, _) = proxy_listener.accept().await.unwrap();
277            let boxed: eggress_core::BoxStream = Box::new(stream);
278            let config = test_config(direct_routing());
279            serve_connection(boxed, config).await
280        });
281
282        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
283        stream.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
284        let mut response = [0u8; 2];
285        stream.read_exact(&mut response).await.unwrap();
286        assert_eq!(response, [0x05, 0x00]);
287
288        stream.write_all(&[0x05, 0x01, 0x00, 0x01]).await.unwrap();
289        match echo_addr.ip() {
290            std::net::IpAddr::V4(ip) => {
291                stream.write_all(&ip.octets()).await.unwrap();
292            }
293            std::net::IpAddr::V6(ip) => {
294                stream.write_all(&ip.octets()).await.unwrap();
295            }
296        }
297        stream
298            .write_all(&echo_addr.port().to_be_bytes())
299            .await
300            .unwrap();
301
302        let mut reply = [0u8; 10];
303        stream.read_exact(&mut reply).await.unwrap();
304        assert_eq!(reply[0], 0x05);
305        assert_eq!(reply[1], 0x00);
306
307        stream.write_all(b"hello").await.unwrap();
308        stream.shutdown().await.unwrap();
309
310        let mut buf = Vec::new();
311        stream.read_to_end(&mut buf).await.unwrap();
312        assert_eq!(&buf, b"hello");
313
314        let report = proxy_jh.await.unwrap();
315        assert!(matches!(report.outcome, execute::SessionOutcome::Completed));
316
317        echo_jh.abort();
318    }
319
320    #[tokio::test]
321    async fn test_serve_connection_http_connect_direct() {
322        let (echo_addr, echo_jh) = eggress_testkit::start_echo_server().await;
323
324        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
325        let proxy_addr = proxy_listener.local_addr().unwrap();
326
327        let _proxy_jh = tokio::spawn(async move {
328            let (stream, _) = proxy_listener.accept().await.unwrap();
329            let boxed: eggress_core::BoxStream = Box::new(stream);
330            let config = test_config(direct_routing());
331            serve_connection(boxed, config).await
332        });
333
334        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
335        let connect_req = format!(
336            "CONNECT {}:{} HTTP/1.1\r\nHost: {}:{}\r\n\r\n",
337            echo_addr.ip(),
338            echo_addr.port(),
339            echo_addr.ip(),
340            echo_addr.port()
341        );
342        stream.write_all(connect_req.as_bytes()).await.unwrap();
343
344        let mut response = vec![0u8; 1024];
345        let n = stream.read(&mut response).await.unwrap();
346        let response_str = String::from_utf8_lossy(&response[..n]);
347        assert!(
348            response_str.contains("200"),
349            "expected 200, got: {response_str}"
350        );
351
352        let header_end = response_str.find("\r\n\r\n").unwrap() + 4;
353        let leftover = &response.as_slice()[header_end..n];
354
355        stream.write_all(b"hello proxy").await.unwrap();
356        stream.shutdown().await.unwrap();
357
358        let mut buf = Vec::new();
359        if !leftover.is_empty() {
360            buf.extend_from_slice(leftover);
361        }
362        stream.read_to_end(&mut buf).await.unwrap();
363        assert_eq!(&buf, b"hello proxy");
364
365        echo_jh.abort();
366    }
367
368    async fn start_echo_origin() -> (std::net::SocketAddr, tokio::task::JoinHandle<()>) {
369        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
370        let addr = listener.local_addr().unwrap();
371
372        let jh = tokio::spawn(async move {
373            loop {
374                let (mut stream, _) = match listener.accept().await {
375                    Ok(s) => s,
376                    Err(_) => break,
377                };
378                tokio::spawn(async move {
379                    use tokio::io::AsyncReadExt;
380                    use tokio::io::AsyncWriteExt;
381
382                    let mut head = Vec::new();
383                    let mut tmp = [0u8; 1];
384                    loop {
385                        if stream.read(&mut tmp).await.unwrap_or(0) == 0 {
386                            return;
387                        }
388                        head.push(tmp[0]);
389                        if head.len() >= 4 && &head[head.len() - 4..] == b"\r\n\r\n" {
390                            break;
391                        }
392                    }
393
394                    let head_str = String::from_utf8_lossy(&head);
395                    let mut content_length: Option<u64> = None;
396                    let mut is_chunked = false;
397                    for line in head_str.lines() {
398                        if let Some((name, value)) = line.split_once(':') {
399                            if name.eq_ignore_ascii_case("Content-Length") {
400                                content_length = match value.trim().parse() {
401                                    Ok(length) => Some(length),
402                                    Err(_) => return,
403                                };
404                            } else if name.eq_ignore_ascii_case("Transfer-Encoding")
405                                && value.trim().eq_ignore_ascii_case("chunked")
406                            {
407                                is_chunked = true;
408                            }
409                        }
410                    }
411
412                    let body = match (content_length, is_chunked) {
413                        (Some(len), _) => {
414                            let mut body = vec![0u8; len as usize];
415                            let mut off = 0;
416                            while off < body.len() {
417                                let n = stream.read(&mut body[off..]).await.unwrap_or(0);
418                                if n == 0 {
419                                    break;
420                                }
421                                off += n;
422                            }
423                            body
424                        }
425                        (None, true) => {
426                            let mut body = Vec::new();
427                            loop {
428                                let mut size_line = Vec::new();
429                                loop {
430                                    let n = stream.read(&mut tmp).await.unwrap_or(0);
431                                    if n == 0 {
432                                        return;
433                                    }
434                                    size_line.push(tmp[0]);
435                                    if size_line.len() >= 2
436                                        && &size_line[size_line.len() - 2..] == b"\r\n"
437                                    {
438                                        break;
439                                    }
440                                }
441                                let size_str =
442                                    String::from_utf8_lossy(&size_line[..size_line.len() - 2]);
443                                let chunk_size = match usize::from_str_radix(size_str.trim(), 16) {
444                                    Ok(size) => size,
445                                    Err(_) => return,
446                                };
447                                if chunk_size == 0 {
448                                    let mut trail = [0u8; 2];
449                                    let _ = stream.read_exact(&mut trail).await;
450                                    break;
451                                }
452                                let mut chunk = vec![0u8; chunk_size];
453                                let mut off = 0;
454                                while off < chunk.len() {
455                                    let n = stream.read(&mut chunk[off..]).await.unwrap_or(0);
456                                    if n == 0 {
457                                        return;
458                                    }
459                                    off += n;
460                                }
461                                body.extend_from_slice(&chunk);
462                                let mut trail = [0u8; 2];
463                                let _ = stream.read_exact(&mut trail).await;
464                            }
465                            body
466                        }
467                        _ => Vec::new(),
468                    };
469
470                    let response = format!(
471                        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
472                        body.len()
473                    );
474                    let _ = stream.write_all(response.as_bytes()).await;
475                    let _ = stream.write_all(&body).await;
476                    let _ = stream.shutdown().await;
477                });
478            }
479        });
480
481        (addr, jh)
482    }
483
484    #[tokio::test]
485    async fn test_http_forward_post_content_length() {
486        let (origin_addr, origin_jh) = start_echo_origin().await;
487
488        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
489        let proxy_addr = proxy_listener.local_addr().unwrap();
490
491        let proxy_jh = tokio::spawn(async move {
492            let (stream, _) = proxy_listener.accept().await.unwrap();
493            let boxed: eggress_core::BoxStream = Box::new(stream);
494            let config = test_config(direct_routing());
495            serve_connection(boxed, config).await
496        });
497
498        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
499        let body = b"hello world";
500        let request = format!(
501            "POST http://{}:{} HTTP/1.1\r\nHost: {}:{}\r\nContent-Length: {}\r\n\r\n",
502            origin_addr.ip(),
503            origin_addr.port(),
504            origin_addr.ip(),
505            origin_addr.port(),
506            body.len()
507        );
508        stream.write_all(request.as_bytes()).await.unwrap();
509        stream.write_all(body).await.unwrap();
510
511        let mut response = Vec::new();
512        stream.read_to_end(&mut response).await.unwrap();
513        let response_str = String::from_utf8_lossy(&response);
514        assert!(
515            response_str.ends_with("hello world"),
516            "body not echoed: {response_str}"
517        );
518
519        let report = proxy_jh.await.unwrap();
520        assert!(matches!(report.outcome, execute::SessionOutcome::Completed));
521
522        origin_jh.abort();
523    }
524
525    #[tokio::test]
526    async fn test_http_forward_post_chunked() {
527        let (origin_addr, origin_jh) = start_echo_origin().await;
528
529        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
530        let proxy_addr = proxy_listener.local_addr().unwrap();
531
532        let proxy_jh = tokio::spawn(async move {
533            let (stream, _) = proxy_listener.accept().await.unwrap();
534            let boxed: eggress_core::BoxStream = Box::new(stream);
535            let config = test_config(direct_routing());
536            serve_connection(boxed, config).await
537        });
538
539        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
540        let body = b"chunked body";
541        let request = format!(
542            "POST http://{}:{} HTTP/1.1\r\nHost: {}:{}\r\nTransfer-Encoding: chunked\r\n\r\n",
543            origin_addr.ip(),
544            origin_addr.port(),
545            origin_addr.ip(),
546            origin_addr.port()
547        );
548        stream.write_all(request.as_bytes()).await.unwrap();
549        stream
550            .write_all(format!("{:x}\r\n", body.len()).as_bytes())
551            .await
552            .unwrap();
553        stream.write_all(body).await.unwrap();
554        stream.write_all(b"\r\n").await.unwrap();
555        stream.write_all(b"0\r\n\r\n").await.unwrap();
556
557        let mut response = Vec::new();
558        stream.read_to_end(&mut response).await.unwrap();
559        let response_str = String::from_utf8_lossy(&response);
560        assert!(
561            response_str.ends_with("chunked body"),
562            "body not echoed: {response_str}"
563        );
564
565        let report = proxy_jh.await.unwrap();
566        assert!(matches!(report.outcome, execute::SessionOutcome::Completed));
567
568        origin_jh.abort();
569    }
570
571    #[tokio::test]
572    async fn test_http_forward_get_no_body() {
573        let (origin_addr, origin_jh) = start_echo_origin().await;
574
575        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
576        let proxy_addr = proxy_listener.local_addr().unwrap();
577
578        let proxy_jh = tokio::spawn(async move {
579            let (stream, _) = proxy_listener.accept().await.unwrap();
580            let boxed: eggress_core::BoxStream = Box::new(stream);
581            let config = test_config(direct_routing());
582            serve_connection(boxed, config).await
583        });
584
585        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
586        let request = format!(
587            "GET http://{}:{}/ HTTP/1.1\r\nHost: {}:{}\r\n\r\n",
588            origin_addr.ip(),
589            origin_addr.port(),
590            origin_addr.ip(),
591            origin_addr.port()
592        );
593        stream.write_all(request.as_bytes()).await.unwrap();
594
595        let mut response = Vec::new();
596        stream.read_to_end(&mut response).await.unwrap();
597        let response_str = String::from_utf8_lossy(&response);
598        assert!(
599            response_str.contains("200 OK"),
600            "expected 200, got: {response_str}"
601        );
602        let body_start = response_str.find("\r\n\r\n").unwrap() + 4;
603        let body = &response_str[body_start..];
604        assert!(body.is_empty(), "expected empty body for GET, got: {body}");
605
606        let report = proxy_jh.await.unwrap();
607        assert!(matches!(report.outcome, execute::SessionOutcome::Completed));
608
609        origin_jh.abort();
610    }
611
612    #[tokio::test(start_paused = true)]
613    async fn test_handshake_timeout_no_bytes() {
614        let (_client_stream, server_stream) = tokio::io::duplex(1024);
615        let boxed: eggress_core::BoxStream = Box::new(server_stream);
616        let config = test_config(direct_routing());
617
618        let task = tokio::spawn(serve_connection(boxed, config));
619
620        tokio::time::advance(Duration::from_secs(6)).await;
621
622        let report = task.await.unwrap();
623        assert!(matches!(
624            report.outcome,
625            execute::SessionOutcome::HandshakeTimedOut
626        ));
627    }
628
629    #[tokio::test(start_paused = true)]
630    async fn test_handshake_timeout_partial_http() {
631        let (mut client_stream, server_stream) = tokio::io::duplex(1024);
632        let boxed: eggress_core::BoxStream = Box::new(server_stream);
633        let config = test_config(direct_routing());
634
635        let task = tokio::spawn(serve_connection(boxed, config));
636
637        client_stream.write_all(b"CON").await.unwrap();
638        tokio::time::advance(Duration::from_secs(6)).await;
639
640        let report = task.await.unwrap();
641        assert!(matches!(
642            report.outcome,
643            execute::SessionOutcome::HandshakeTimedOut
644        ));
645    }
646
647    #[tokio::test(start_paused = true)]
648    async fn test_handshake_timeout_partial_socks5() {
649        let (mut client_stream, server_stream) = tokio::io::duplex(1024);
650        let boxed: eggress_core::BoxStream = Box::new(server_stream);
651        let config = test_config(direct_routing());
652
653        let task = tokio::spawn(serve_connection(boxed, config));
654
655        client_stream.write_all(&[0x05]).await.unwrap();
656        tokio::time::advance(Duration::from_secs(6)).await;
657
658        let report = task.await.unwrap();
659        assert!(matches!(
660            report.outcome,
661            execute::SessionOutcome::HandshakeTimedOut
662        ));
663    }
664
665    #[tokio::test]
666    async fn test_handshake_completes_before_timeout() {
667        let (echo_addr, echo_jh) = eggress_testkit::start_echo_server().await;
668
669        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
670        let proxy_addr = proxy_listener.local_addr().unwrap();
671
672        let proxy_jh = tokio::spawn(async move {
673            let (stream, _) = proxy_listener.accept().await.unwrap();
674            let boxed: eggress_core::BoxStream = Box::new(stream);
675            let config = test_config(direct_routing());
676            serve_connection(boxed, config).await
677        });
678
679        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
680        stream.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
681        let mut response = [0u8; 2];
682        stream.read_exact(&mut response).await.unwrap();
683        assert_eq!(response, [0x05, 0x00]);
684
685        stream.write_all(&[0x05, 0x01, 0x00, 0x01]).await.unwrap();
686        match echo_addr.ip() {
687            std::net::IpAddr::V4(ip) => {
688                stream.write_all(&ip.octets()).await.unwrap();
689            }
690            std::net::IpAddr::V6(ip) => {
691                stream.write_all(&ip.octets()).await.unwrap();
692            }
693        }
694        stream
695            .write_all(&echo_addr.port().to_be_bytes())
696            .await
697            .unwrap();
698
699        let mut reply = [0u8; 10];
700        stream.read_exact(&mut reply).await.unwrap();
701        assert_eq!(reply[0], 0x05);
702        assert_eq!(reply[1], 0x00);
703
704        stream.write_all(b"hello").await.unwrap();
705        stream.shutdown().await.unwrap();
706
707        let mut buf = Vec::new();
708        stream.read_to_end(&mut buf).await.unwrap();
709        assert_eq!(&buf, b"hello");
710
711        let report = proxy_jh.await.unwrap();
712        assert!(matches!(report.outcome, execute::SessionOutcome::Completed));
713
714        echo_jh.abort();
715    }
716
717    #[tokio::test]
718    async fn test_http_forward_get_reports_nonzero_bytes() {
719        let (origin_addr, origin_jh) = start_echo_origin().await;
720
721        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
722        let proxy_addr = proxy_listener.local_addr().unwrap();
723
724        let proxy_jh = tokio::spawn(async move {
725            let (stream, _) = proxy_listener.accept().await.unwrap();
726            let boxed: eggress_core::BoxStream = Box::new(stream);
727            let config = test_config(direct_routing());
728            serve_connection(boxed, config).await
729        });
730
731        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
732        let request = format!(
733            "GET http://{}:{}/ HTTP/1.1\r\nHost: {}:{}\r\n\r\n",
734            origin_addr.ip(),
735            origin_addr.port(),
736            origin_addr.ip(),
737            origin_addr.port()
738        );
739        stream.write_all(request.as_bytes()).await.unwrap();
740
741        let mut response = Vec::new();
742        stream.read_to_end(&mut response).await.unwrap();
743
744        let report = proxy_jh.await.unwrap();
745        assert!(matches!(report.outcome, execute::SessionOutcome::Completed));
746        assert!(
747            report.bytes_upstream > 0,
748            "upstream bytes should be nonzero"
749        );
750        assert!(
751            report.bytes_downstream > 0,
752            "downstream bytes should be nonzero"
753        );
754
755        origin_jh.abort();
756    }
757
758    #[tokio::test]
759    async fn test_http_forward_post_reports_body_bytes() {
760        let (origin_addr, origin_jh) = start_echo_origin().await;
761
762        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
763        let proxy_addr = proxy_listener.local_addr().unwrap();
764
765        let proxy_jh = tokio::spawn(async move {
766            let (stream, _) = proxy_listener.accept().await.unwrap();
767            let boxed: eggress_core::BoxStream = Box::new(stream);
768            let config = test_config(direct_routing());
769            serve_connection(boxed, config).await
770        });
771
772        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
773        let body = b"hello world";
774        let request = format!(
775            "POST http://{}:{} HTTP/1.1\r\nHost: {}:{}\r\nContent-Length: {}\r\n\r\n",
776            origin_addr.ip(),
777            origin_addr.port(),
778            origin_addr.ip(),
779            origin_addr.port(),
780            body.len()
781        );
782        stream.write_all(request.as_bytes()).await.unwrap();
783        stream.write_all(body).await.unwrap();
784
785        let mut response = Vec::new();
786        stream.read_to_end(&mut response).await.unwrap();
787
788        let report = proxy_jh.await.unwrap();
789        assert!(matches!(report.outcome, execute::SessionOutcome::Completed));
790        assert!(
791            report.bytes_upstream > body.len() as u64,
792            "upstream bytes ({}) should exceed body length ({})",
793            report.bytes_upstream,
794            body.len()
795        );
796        assert!(report.bytes_downstream > 0);
797
798        origin_jh.abort();
799    }
800
801    #[tokio::test]
802    async fn test_successful_session_has_no_failure() {
803        let (echo_addr, echo_jh) = eggress_testkit::start_echo_server().await;
804
805        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
806        let proxy_addr = proxy_listener.local_addr().unwrap();
807
808        let proxy_jh = tokio::spawn(async move {
809            let (stream, _) = proxy_listener.accept().await.unwrap();
810            let boxed: eggress_core::BoxStream = Box::new(stream);
811            let config = test_config(direct_routing());
812            serve_connection(boxed, config).await
813        });
814
815        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
816        stream.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
817        let mut response = [0u8; 2];
818        stream.read_exact(&mut response).await.unwrap();
819        assert_eq!(response, [0x05, 0x00]);
820
821        stream.write_all(&[0x05, 0x01, 0x00, 0x01]).await.unwrap();
822        match echo_addr.ip() {
823            std::net::IpAddr::V4(ip) => {
824                stream.write_all(&ip.octets()).await.unwrap();
825            }
826            std::net::IpAddr::V6(ip) => {
827                stream.write_all(&ip.octets()).await.unwrap();
828            }
829        }
830        stream
831            .write_all(&echo_addr.port().to_be_bytes())
832            .await
833            .unwrap();
834
835        let mut reply = [0u8; 10];
836        stream.read_exact(&mut reply).await.unwrap();
837        assert_eq!(reply[0], 0x05);
838        assert_eq!(reply[1], 0x00);
839
840        stream.write_all(b"hello").await.unwrap();
841        stream.shutdown().await.unwrap();
842
843        let mut buf = Vec::new();
844        stream.read_to_end(&mut buf).await.unwrap();
845        assert_eq!(&buf, b"hello");
846
847        let report = proxy_jh.await.unwrap();
848        assert!(matches!(report.outcome, execute::SessionOutcome::Completed));
849        assert_eq!(report.failure, None);
850
851        echo_jh.abort();
852    }
853
854    #[tokio::test(start_paused = true)]
855    async fn test_handshake_timeout_maps_to_failure_category() {
856        let (_client_stream, server_stream) = tokio::io::duplex(1024);
857        let boxed: eggress_core::BoxStream = Box::new(server_stream);
858        let config = test_config(direct_routing());
859
860        let task = tokio::spawn(serve_connection(boxed, config));
861
862        tokio::time::advance(Duration::from_secs(6)).await;
863
864        let report = task.await.unwrap();
865        assert!(matches!(
866            report.outcome,
867            execute::SessionOutcome::HandshakeTimedOut
868        ));
869        assert_eq!(
870            report.failure,
871            Some(execute::FailureCategory::HandshakeTimeout)
872        );
873    }
874
875    #[tokio::test]
876    async fn test_failure_category_from_session_open_error_dns() {
877        let error = SessionOpenError::Dns;
878        let category = execute::FailureCategory::from(&error);
879        assert_eq!(category, execute::FailureCategory::Dns);
880    }
881
882    #[tokio::test]
883    async fn test_failure_category_from_session_open_error_refused() {
884        let error = SessionOpenError::Refused;
885        let category = execute::FailureCategory::from(&error);
886        assert_eq!(category, execute::FailureCategory::ConnectionRefused);
887    }
888
889    #[tokio::test]
890    async fn test_failure_category_from_session_open_error_network_unreachable() {
891        let error = SessionOpenError::NetworkUnreachable;
892        let category = execute::FailureCategory::from(&error);
893        assert_eq!(category, execute::FailureCategory::NetworkUnreachable);
894    }
895
896    #[tokio::test]
897    async fn test_failure_category_from_session_open_error_host_unreachable() {
898        let error = SessionOpenError::HostUnreachable;
899        let category = execute::FailureCategory::from(&error);
900        assert_eq!(category, execute::FailureCategory::HostUnreachable);
901    }
902
903    #[tokio::test]
904    async fn test_failure_category_from_session_open_error_timeout() {
905        let error = SessionOpenError::Timeout;
906        let category = execute::FailureCategory::from(&error);
907        assert_eq!(category, execute::FailureCategory::RouteTimeout);
908    }
909
910    #[tokio::test]
911    async fn test_failure_category_from_session_open_error_upstream_auth() {
912        let error = SessionOpenError::UpstreamAuthentication;
913        let category = execute::FailureCategory::from(&error);
914        assert_eq!(category, execute::FailureCategory::UpstreamAuthentication);
915    }
916
917    #[tokio::test]
918    async fn test_failure_category_from_io_error_connection_refused() {
919        let error = std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "refused");
920        let category = execute::FailureCategory::from_io_error(&error);
921        assert_eq!(category, execute::FailureCategory::ConnectionRefused);
922    }
923
924    #[tokio::test]
925    async fn test_failure_category_from_io_error_connection_reset() {
926        let error = std::io::Error::new(std::io::ErrorKind::ConnectionReset, "reset");
927        let category = execute::FailureCategory::from_io_error(&error);
928        assert_eq!(category, execute::FailureCategory::Relay);
929    }
930
931    #[tokio::test]
932    async fn test_failure_category_from_io_error_timeout() {
933        let error = std::io::Error::new(std::io::ErrorKind::TimedOut, "timeout");
934        let category = execute::FailureCategory::from_io_error(&error);
935        assert_eq!(category, execute::FailureCategory::Relay);
936    }
937
938    #[tokio::test]
939    async fn test_route_failure_maps_to_dns_category() {
940        let report = execute::SessionReport::open_failed(
941            SessionOpenError::Dns,
942            Some("socks5".to_string()),
943            Some("example.com:443".to_string()),
944            "direct".to_string(),
945        );
946        assert!(matches!(
947            report.outcome,
948            execute::SessionOutcome::RouteFailed
949        ));
950        assert_eq!(report.failure, Some(execute::FailureCategory::Dns));
951    }
952
953    #[tokio::test]
954    async fn test_route_failure_maps_to_connection_refused_category() {
955        let report = execute::SessionReport::open_failed(
956            SessionOpenError::Refused,
957            Some("http".to_string()),
958            Some("10.0.0.1:80".to_string()),
959            "chain(2)".to_string(),
960        );
961        assert!(matches!(
962            report.outcome,
963            execute::SessionOutcome::RouteFailed
964        ));
965        assert_eq!(
966            report.failure,
967            Some(execute::FailureCategory::ConnectionRefused)
968        );
969    }
970
971    #[tokio::test]
972    async fn test_completed_session_has_no_failure() {
973        let report = execute::SessionReport::completed(
974            Some("socks5".to_string()),
975            Some("example.com:443".to_string()),
976            "direct".to_string(),
977            1024,
978            2048,
979        );
980        assert!(matches!(report.outcome, execute::SessionOutcome::Completed));
981        assert_eq!(report.failure, None);
982        assert_eq!(report.bytes_upstream, 1024);
983        assert_eq!(report.bytes_downstream, 2048);
984    }
985
986    #[tokio::test]
987    async fn test_cancelled_session_has_cancelled_failure() {
988        let report = execute::SessionReport::cancelled(
989            Some("http".to_string()),
990            Some("example.com:80".to_string()),
991            "direct".to_string(),
992        );
993        assert!(matches!(report.outcome, execute::SessionOutcome::Cancelled));
994        assert_eq!(report.failure, Some(execute::FailureCategory::Cancelled));
995    }
996
997    #[tokio::test]
998    async fn test_authentication_failure_maps_to_failure_category() {
999        let auth = accept::InboundAuthentication::UsernamePassword {
1000            username: "user".to_string(),
1001            password: "secret".to_string(),
1002        };
1003
1004        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1005        let proxy_addr = proxy_listener.local_addr().unwrap();
1006
1007        let proxy_jh = tokio::spawn(async move {
1008            let (stream, _) = proxy_listener.accept().await.unwrap();
1009            let boxed: eggress_core::BoxStream = Box::new(stream);
1010            let mut cfg = test_config(direct_routing());
1011            cfg.authentication = auth;
1012            let config = cfg;
1013            serve_connection(boxed, config).await
1014        });
1015
1016        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
1017        stream.write_all(&[0x05, 0x02, 0x00, 0x02]).await.unwrap();
1018        let mut response = [0u8; 2];
1019        stream.read_exact(&mut response).await.unwrap();
1020        assert_eq!(response, [0x05, 0x02]);
1021
1022        stream
1023            .write_all(&[0x01, 0x04, b'u', b's', b'e', b'r', 0x05])
1024            .await
1025            .unwrap();
1026        stream.write_all(b"wrong").await.unwrap();
1027        let mut auth_resp = [0u8; 2];
1028        stream.read_exact(&mut auth_resp).await.unwrap();
1029        assert_eq!(auth_resp, [0x01, 0x01]);
1030
1031        let report = proxy_jh.await.unwrap();
1032        assert!(matches!(
1033            report.outcome,
1034            execute::SessionOutcome::AuthenticationFailed
1035        ));
1036        assert_eq!(
1037            report.failure,
1038            Some(execute::FailureCategory::Authentication)
1039        );
1040    }
1041
1042    #[tokio::test]
1043    async fn test_reject_route_returns_403_for_http() {
1044        let rules = vec![eggress_routing::CompiledRule {
1045            id: eggress_routing::RuleId(std::sync::Arc::from("block")),
1046            matcher: eggress_routing::MatchExpr::Any,
1047            action: eggress_routing::RouteActionSpec::Reject(
1048                eggress_core::RejectReason::AccessDenied,
1049            ),
1050        }];
1051        let routing: Arc<dyn RouteService> =
1052            Arc::new(Router::new(rules, eggress_routing::RouteActionSpec::Direct));
1053
1054        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1055        let proxy_addr = proxy_listener.local_addr().unwrap();
1056
1057        let _proxy_jh = tokio::spawn(async move {
1058            let (stream, _) = proxy_listener.accept().await.unwrap();
1059            let boxed: eggress_core::BoxStream = Box::new(stream);
1060            let config = test_config(routing);
1061            serve_connection(boxed, config).await
1062        });
1063
1064        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
1065        let request = "GET http://example.com/ HTTP/1.1\r\nHost: example.com\r\n\r\n";
1066        stream.write_all(request.as_bytes()).await.unwrap();
1067
1068        let mut response = Vec::new();
1069        stream.read_to_end(&mut response).await.unwrap();
1070        let response_str = String::from_utf8_lossy(&response);
1071        assert!(
1072            response_str.contains("403"),
1073            "expected 403, got: {response_str}"
1074        );
1075    }
1076
1077    #[tokio::test]
1078    async fn test_source_cidr_matching_with_real_peer() {
1079        let (echo_addr, echo_jh) = eggress_testkit::start_echo_server().await;
1080
1081        let rules = vec![eggress_routing::CompiledRule {
1082            id: eggress_routing::RuleId(std::sync::Arc::from("allow-localhost")),
1083            matcher: eggress_routing::MatchExpr::SourceCidr("127.0.0.0/8".parse().unwrap()),
1084            action: eggress_routing::RouteActionSpec::Direct,
1085        }];
1086        let routing: Arc<dyn RouteService> =
1087            Arc::new(Router::new(rules, eggress_routing::RouteActionSpec::Direct));
1088
1089        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1090        let proxy_addr = proxy_listener.local_addr().unwrap();
1091
1092        let proxy_jh = tokio::spawn(async move {
1093            let (stream, peer) = proxy_listener.accept().await.unwrap();
1094            let boxed: eggress_core::BoxStream = Box::new(stream);
1095            let mut cfg = test_config(routing.clone());
1096            cfg.context = ConnectionContext {
1097                source: Some(peer),
1098                listener: "test-listener".to_string(),
1099                generation: 0,
1100            };
1101            let config = cfg;
1102            serve_connection(boxed, config).await
1103        });
1104
1105        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
1106        stream.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
1107        let mut response = [0u8; 2];
1108        stream.read_exact(&mut response).await.unwrap();
1109        assert_eq!(response, [0x05, 0x00]);
1110
1111        stream.write_all(&[0x05, 0x01, 0x00, 0x01]).await.unwrap();
1112        match echo_addr.ip() {
1113            std::net::IpAddr::V4(ip) => {
1114                stream.write_all(&ip.octets()).await.unwrap();
1115            }
1116            std::net::IpAddr::V6(ip) => {
1117                stream.write_all(&ip.octets()).await.unwrap();
1118            }
1119        }
1120        stream
1121            .write_all(&echo_addr.port().to_be_bytes())
1122            .await
1123            .unwrap();
1124
1125        let mut reply = [0u8; 10];
1126        stream.read_exact(&mut reply).await.unwrap();
1127        assert_eq!(reply[0], 0x05);
1128        assert_eq!(reply[1], 0x00);
1129
1130        stream.write_all(b"hello").await.unwrap();
1131        stream.shutdown().await.unwrap();
1132
1133        let mut buf = Vec::new();
1134        stream.read_to_end(&mut buf).await.unwrap();
1135        assert_eq!(&buf, b"hello");
1136
1137        let report = proxy_jh.await.unwrap();
1138        assert!(matches!(report.outcome, execute::SessionOutcome::Completed));
1139
1140        echo_jh.abort();
1141    }
1142
1143    #[tokio::test]
1144    async fn test_http_expectation_is_rejected_with_417_and_connection_close() {
1145        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1146        let proxy_addr = proxy_listener.local_addr().unwrap();
1147
1148        let proxy_jh = tokio::spawn(async move {
1149            let (stream, _) = proxy_listener.accept().await.unwrap();
1150            let boxed: eggress_core::BoxStream = Box::new(stream);
1151            let config = test_config(direct_routing());
1152            serve_connection(boxed, config).await
1153        });
1154
1155        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
1156        stream
1157            .write_all(
1158                b"POST http://127.0.0.1:1/ HTTP/1.1\r\nHost: 127.0.0.1:1\r\nExpect: 100-continue\r\nContent-Length: 1048576\r\n\r\n",
1159            )
1160            .await
1161            .unwrap();
1162
1163        let mut response = Vec::new();
1164        tokio::time::timeout(Duration::from_secs(1), stream.read_to_end(&mut response))
1165            .await
1166            .expect("expectation rejection must be bounded")
1167            .unwrap();
1168        assert!(String::from_utf8_lossy(&response).starts_with("HTTP/1.1 417 Expectation Failed"));
1169
1170        let report = proxy_jh.await.unwrap();
1171        assert!(matches!(
1172            report.outcome,
1173            execute::SessionOutcome::ClientProtocolError
1174        ));
1175    }
1176
1177    #[tokio::test]
1178    async fn test_http_body_upload_is_not_limited_by_connect_timeout() {
1179        let origin_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1180        let origin_addr = origin_listener.local_addr().unwrap();
1181        let origin_jh = tokio::spawn(async move {
1182            let (stream, _) = origin_listener.accept().await.unwrap();
1183            let mut reader = BufReader::new(stream);
1184            let mut line = Vec::new();
1185            loop {
1186                line.clear();
1187                reader.read_until(b'\n', &mut line).await.unwrap();
1188                if line == b"\r\n" {
1189                    break;
1190                }
1191            }
1192            let mut body = [0u8; 16];
1193            reader.read_exact(&mut body).await.unwrap();
1194            let mut stream = reader.into_inner();
1195            stream
1196                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
1197                .await
1198                .unwrap();
1199        });
1200
1201        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1202        let proxy_addr = proxy_listener.local_addr().unwrap();
1203        let proxy_jh = tokio::spawn(async move {
1204            let (stream, _) = proxy_listener.accept().await.unwrap();
1205            let boxed: eggress_core::BoxStream = Box::new(stream);
1206            let mut config = test_config(direct_routing());
1207            config.connect_timeout = Duration::from_millis(50);
1208            serve_connection(boxed, config).await
1209        });
1210
1211        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
1212        let request = format!(
1213            "POST http://{}:{}/ HTTP/1.1\r\nHost: {}:{}\r\nContent-Length: 16\r\n\r\n",
1214            origin_addr.ip(),
1215            origin_addr.port(),
1216            origin_addr.ip(),
1217            origin_addr.port()
1218        );
1219        stream.write_all(request.as_bytes()).await.unwrap();
1220        tokio::time::sleep(Duration::from_millis(100)).await;
1221        stream.write_all(b"delayed-body-123").await.unwrap();
1222
1223        let mut response = Vec::new();
1224        tokio::time::timeout(Duration::from_secs(1), stream.read_to_end(&mut response))
1225            .await
1226            .expect("body upload must not inherit the connect timeout")
1227            .unwrap();
1228        let report = proxy_jh.await.unwrap();
1229        assert!(matches!(report.outcome, execute::SessionOutcome::Completed));
1230
1231        origin_jh.abort();
1232    }
1233}
1234
1235/// Tests proving that session metrics are structurally balanced: one
1236/// `record_session_start()` followed by exactly one `record_session()` for
1237/// every path through `serve_connection()`.
1238#[cfg(test)]
1239mod metrics_lifecycle_tests {
1240    use super::*;
1241    use eggress_routing::{RouteActionSpec, Router};
1242    use std::sync::atomic::{AtomicUsize, Ordering};
1243    use std::sync::Mutex;
1244    use tokio::io::{AsyncReadExt, AsyncWriteExt};
1245
1246    /// A test-only metrics implementation that counts calls.
1247    struct RecordingMetrics {
1248        starts: AtomicUsize,
1249        terminals: AtomicUsize,
1250        auth_failures: AtomicUsize,
1251        terminal_reports: Mutex<Vec<execute::SessionOutcome>>,
1252    }
1253
1254    impl RecordingMetrics {
1255        fn new() -> Self {
1256            Self {
1257                starts: AtomicUsize::new(0),
1258                terminals: AtomicUsize::new(0),
1259                auth_failures: AtomicUsize::new(0),
1260                terminal_reports: Mutex::new(Vec::new()),
1261            }
1262        }
1263    }
1264
1265    impl SessionMetrics for RecordingMetrics {
1266        fn record_session_start(&self) {
1267            self.starts.fetch_add(1, Ordering::SeqCst);
1268        }
1269        fn record_session(&self, report: &SessionReport) {
1270            self.terminals.fetch_add(1, Ordering::SeqCst);
1271            self.terminal_reports
1272                .lock()
1273                .unwrap()
1274                .push(std::mem::replace(
1275                    &mut report.outcome.clone_outcome(),
1276                    execute::SessionOutcome::Completed,
1277                ));
1278        }
1279        fn record_auth_failure(&self) {
1280            self.auth_failures.fetch_add(1, Ordering::SeqCst);
1281        }
1282        fn record_route_decision(&self, _: &str, _: &str, _: &str) {}
1283        fn record_upstream_open(&self, _: &str, _: &str) {}
1284        fn record_upstream_failure(&self, _: &str, _: &str) {}
1285    }
1286
1287    impl execute::SessionOutcome {
1288        fn clone_outcome(&self) -> Self {
1289            match self {
1290                Self::Completed => Self::Completed,
1291                Self::ClientProtocolError => Self::ClientProtocolError,
1292                Self::AuthenticationFailed => Self::AuthenticationFailed,
1293                Self::HandshakeTimedOut => Self::HandshakeTimedOut,
1294                Self::RouteFailed => Self::RouteFailed,
1295                Self::RelayFailed => Self::RelayFailed,
1296                Self::Cancelled => Self::Cancelled,
1297            }
1298        }
1299    }
1300
1301    fn test_direct_routing() -> Arc<dyn RouteService> {
1302        Arc::new(Router::new(vec![], RouteActionSpec::Direct))
1303    }
1304
1305    fn test_all_protocols() -> Arc<[eggress_core::ProtocolId]> {
1306        Arc::from([
1307            eggress_core::ProtocolId::Http,
1308            eggress_core::ProtocolId::Socks4,
1309            eggress_core::ProtocolId::Socks5,
1310            eggress_core::ProtocolId::Http2,
1311            eggress_core::ProtocolId::WebSocket,
1312            eggress_core::ProtocolId::Raw,
1313        ])
1314    }
1315
1316    fn metrics_config(metrics: Arc<RecordingMetrics>) -> ConnectionConfig {
1317        ConnectionConfig {
1318            routing: test_direct_routing(),
1319            context: ConnectionContext::default(),
1320            handshake_timeout: Duration::from_secs(5),
1321            connect_timeout: Duration::from_secs(10),
1322            protocols: test_all_protocols(),
1323            authentication: accept::InboundAuthentication::None,
1324            metrics: Some(metrics),
1325            udp: None,
1326            tls_client_config: None,
1327            shadowsocks: None,
1328            shadowsocks_metrics: None,
1329            trojan: None,
1330            fixed_target: None,
1331            local_bind: None,
1332        }
1333    }
1334
1335    #[tokio::test]
1336    async fn metrics_balanced_after_successful_session() {
1337        let (echo_addr, echo_jh) = eggress_testkit::start_echo_server().await;
1338        let metrics = Arc::new(RecordingMetrics::new());
1339        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1340        let proxy_addr = proxy_listener.local_addr().unwrap();
1341        let m = metrics.clone();
1342        let proxy_jh = tokio::spawn(async move {
1343            let (stream, _) = proxy_listener.accept().await.unwrap();
1344            let boxed: eggress_core::BoxStream = Box::new(stream);
1345            let config = metrics_config(m);
1346            serve_connection(boxed, config).await
1347        });
1348
1349        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
1350        stream.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
1351        let mut response = [0u8; 2];
1352        stream.read_exact(&mut response).await.unwrap();
1353        assert_eq!(response, [0x05, 0x00]);
1354
1355        stream.write_all(&[0x05, 0x01, 0x00, 0x01]).await.unwrap();
1356        match echo_addr.ip() {
1357            std::net::IpAddr::V4(ip) => stream.write_all(&ip.octets()).await.unwrap(),
1358            std::net::IpAddr::V6(ip) => stream.write_all(&ip.octets()).await.unwrap(),
1359        }
1360        stream
1361            .write_all(&echo_addr.port().to_be_bytes())
1362            .await
1363            .unwrap();
1364
1365        let mut reply = [0u8; 10];
1366        stream.read_exact(&mut reply).await.unwrap();
1367        stream.write_all(b"hello").await.unwrap();
1368        stream.shutdown().await.unwrap();
1369        let mut buf = Vec::new();
1370        stream.read_to_end(&mut buf).await.unwrap();
1371
1372        let report = proxy_jh.await.unwrap();
1373        assert!(matches!(report.outcome, execute::SessionOutcome::Completed));
1374
1375        let m = &*metrics;
1376        assert_eq!(m.starts.load(Ordering::SeqCst), 1, "expected 1 start");
1377        assert_eq!(m.terminals.load(Ordering::SeqCst), 1, "expected 1 terminal");
1378        assert_eq!(
1379            m.auth_failures.load(Ordering::SeqCst),
1380            0,
1381            "no auth failures"
1382        );
1383        echo_jh.abort();
1384    }
1385
1386    #[tokio::test]
1387    async fn metrics_balanced_after_auth_failure() {
1388        let auth = accept::InboundAuthentication::UsernamePassword {
1389            username: "user".to_string(),
1390            password: "secret".to_string(),
1391        };
1392        let metrics = Arc::new(RecordingMetrics::new());
1393        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1394        let proxy_addr = proxy_listener.local_addr().unwrap();
1395        let m = metrics.clone();
1396        let proxy_jh = tokio::spawn(async move {
1397            let (stream, _) = proxy_listener.accept().await.unwrap();
1398            let boxed: eggress_core::BoxStream = Box::new(stream);
1399            let mut cfg = metrics_config(m);
1400            cfg.authentication = auth;
1401            serve_connection(boxed, cfg).await
1402        });
1403
1404        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
1405        stream.write_all(&[0x05, 0x02, 0x00, 0x02]).await.unwrap();
1406        let mut response = [0u8; 2];
1407        stream.read_exact(&mut response).await.unwrap();
1408        assert_eq!(response, [0x05, 0x02]);
1409
1410        stream
1411            .write_all(&[0x01, 0x04, b'u', b's', b'e', b'r', 0x05])
1412            .await
1413            .unwrap();
1414        stream.write_all(b"wrong").await.unwrap();
1415        let mut auth_resp = [0u8; 2];
1416        stream.read_exact(&mut auth_resp).await.unwrap();
1417
1418        let report = proxy_jh.await.unwrap();
1419        assert!(matches!(
1420            report.outcome,
1421            execute::SessionOutcome::AuthenticationFailed
1422        ));
1423
1424        let m = &*metrics;
1425        assert_eq!(m.starts.load(Ordering::SeqCst), 1);
1426        assert_eq!(m.terminals.load(Ordering::SeqCst), 1);
1427        assert_eq!(m.auth_failures.load(Ordering::SeqCst), 1);
1428    }
1429
1430    #[tokio::test]
1431    async fn metrics_balanced_after_protocol_error() {
1432        let metrics = Arc::new(RecordingMetrics::new());
1433        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1434        let proxy_addr = proxy_listener.local_addr().unwrap();
1435        let m = metrics.clone();
1436        let proxy_jh = tokio::spawn(async move {
1437            let (stream, _) = proxy_listener.accept().await.unwrap();
1438            let boxed: eggress_core::BoxStream = Box::new(stream);
1439            serve_connection(boxed, metrics_config(m)).await
1440        });
1441
1442        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
1443        stream.write_all(b"garbage data").await.unwrap();
1444        stream.shutdown().await.unwrap();
1445
1446        let report = proxy_jh.await.unwrap();
1447        assert!(matches!(
1448            report.outcome,
1449            execute::SessionOutcome::ClientProtocolError
1450        ));
1451
1452        let m = &*metrics;
1453        assert_eq!(m.starts.load(Ordering::SeqCst), 1);
1454        assert_eq!(m.terminals.load(Ordering::SeqCst), 1);
1455        assert_eq!(m.auth_failures.load(Ordering::SeqCst), 0);
1456    }
1457
1458    #[tokio::test(start_paused = true)]
1459    async fn metrics_balanced_after_handshake_timeout() {
1460        let metrics = Arc::new(RecordingMetrics::new());
1461        let (_client_stream, server_stream) = tokio::io::duplex(1024);
1462        let boxed: eggress_core::BoxStream = Box::new(server_stream);
1463        let task = tokio::spawn(serve_connection(boxed, metrics_config(metrics.clone())));
1464        tokio::time::advance(Duration::from_secs(6)).await;
1465
1466        let report = task.await.unwrap();
1467        assert!(matches!(
1468            report.outcome,
1469            execute::SessionOutcome::HandshakeTimedOut
1470        ));
1471
1472        let m = &*metrics;
1473        assert_eq!(m.starts.load(Ordering::SeqCst), 1);
1474        assert_eq!(m.terminals.load(Ordering::SeqCst), 1);
1475        assert_eq!(m.auth_failures.load(Ordering::SeqCst), 0);
1476    }
1477
1478    #[tokio::test]
1479    async fn no_double_finalization_for_route_failure() {
1480        let rules = vec![eggress_routing::CompiledRule {
1481            id: eggress_routing::RuleId(std::sync::Arc::from("block")),
1482            matcher: eggress_routing::MatchExpr::Any,
1483            action: eggress_routing::RouteActionSpec::Reject(
1484                eggress_core::RejectReason::AccessDenied,
1485            ),
1486        }];
1487        let routing: Arc<dyn eggress_routing::RouteService> =
1488            Arc::new(Router::new(rules, eggress_routing::RouteActionSpec::Direct));
1489        let metrics = Arc::new(RecordingMetrics::new());
1490        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1491        let proxy_addr = proxy_listener.local_addr().unwrap();
1492        let m = metrics.clone();
1493        let proxy_jh = tokio::spawn(async move {
1494            let (stream, _) = proxy_listener.accept().await.unwrap();
1495            let boxed: eggress_core::BoxStream = Box::new(stream);
1496            let mut cfg = metrics_config(m);
1497            cfg.routing = routing;
1498            serve_connection(boxed, cfg).await
1499        });
1500
1501        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
1502        let request = "GET http://example.com/ HTTP/1.1\r\nHost: example.com\r\n\r\n";
1503        stream.write_all(request.as_bytes()).await.unwrap();
1504        let mut response = Vec::new();
1505        stream.read_to_end(&mut response).await.unwrap();
1506
1507        let report = proxy_jh.await.unwrap();
1508        assert!(matches!(
1509            report.outcome,
1510            execute::SessionOutcome::RouteFailed | execute::SessionOutcome::ClientProtocolError
1511        ));
1512
1513        let m = &*metrics;
1514        assert_eq!(m.starts.load(Ordering::SeqCst), 1);
1515        assert_eq!(m.terminals.load(Ordering::SeqCst), 1);
1516    }
1517}
1518
1519/// Negative tests for lean (common-only) build.
1520///
1521/// These tests verify that excluded protocol capabilities fail clearly
1522/// at the accept boundary, never silently degrading. They run only when
1523/// the `extended` feature is disabled (lean build).
1524#[cfg(all(test, not(feature = "extended")))]
1525mod lean_negative_tests {
1526    use super::*;
1527    use eggress_core::ProtocolId;
1528    use tokio::io::{AsyncReadExt, AsyncWriteExt};
1529
1530    #[tokio::test]
1531    async fn lean_rejects_shadowsocks_accept() {
1532        use eggress_routing::{RouteActionSpec, Router};
1533
1534        let routing: Arc<dyn eggress_routing::RouteService> =
1535            Arc::new(Router::new(vec![], RouteActionSpec::Direct));
1536        let protocols: Arc<[ProtocolId]> = Arc::from([ProtocolId::Shadowsocks]);
1537        let (_client, server) = tokio::io::duplex(1024);
1538        let boxed: eggress_core::BoxStream = Box::new(server);
1539
1540        let config = ConnectionConfig {
1541            routing,
1542            context: ConnectionContext::default(),
1543            handshake_timeout: Duration::from_secs(2),
1544            connect_timeout: Duration::from_secs(5),
1545            protocols,
1546            authentication: accept::InboundAuthentication::None,
1547            metrics: None,
1548            udp: None,
1549            tls_client_config: None,
1550            shadowsocks: Some(accept::InboundShadowsocksConfig {
1551                method: "aes-256-gcm".to_string(),
1552                password: "test-password".to_string(),
1553            }),
1554            shadowsocks_metrics: None,
1555            trojan: None,
1556            fixed_target: None,
1557            local_bind: None,
1558        };
1559
1560        let report = serve_connection(boxed, config).await;
1561        // In lean build, shadowsocks accept should fail with a protocol error
1562        // because the feature is not included.
1563        assert!(
1564            matches!(
1565                report.outcome,
1566                execute::SessionOutcome::ClientProtocolError
1567                    | execute::SessionOutcome::HandshakeTimedOut
1568            ),
1569            "shadowsocks should fail in lean build, got: {:?}",
1570            report.outcome
1571        );
1572    }
1573
1574    #[tokio::test]
1575    async fn lean_rejects_trojan_accept() {
1576        use eggress_routing::{RouteActionSpec, Router};
1577
1578        let routing: Arc<dyn eggress_routing::RouteService> =
1579            Arc::new(Router::new(vec![], RouteActionSpec::Direct));
1580        let protocols: Arc<[ProtocolId]> = Arc::from([ProtocolId::Trojan]);
1581        let (_client, server) = tokio::io::duplex(1024);
1582        let boxed: eggress_core::BoxStream = Box::new(server);
1583
1584        let config = ConnectionConfig {
1585            routing,
1586            context: ConnectionContext::default(),
1587            handshake_timeout: Duration::from_secs(2),
1588            connect_timeout: Duration::from_secs(5),
1589            protocols,
1590            authentication: accept::InboundAuthentication::None,
1591            metrics: None,
1592            udp: None,
1593            tls_client_config: None,
1594            shadowsocks: None,
1595            shadowsocks_metrics: None,
1596            trojan: Some(accept::InboundTrojanConfig {
1597                password: "test-password".to_string(),
1598                fallback: None,
1599            }),
1600            fixed_target: None,
1601            local_bind: None,
1602        };
1603
1604        let report = serve_connection(boxed, config).await;
1605        // In lean build, trojan accept should fail with a protocol error
1606        // because the feature is not included.
1607        assert!(
1608            matches!(
1609                report.outcome,
1610                execute::SessionOutcome::ClientProtocolError
1611                    | execute::SessionOutcome::HandshakeTimedOut
1612            ),
1613            "trojan should fail in lean build, got: {:?}",
1614            report.outcome
1615        );
1616    }
1617
1618    #[tokio::test]
1619    async fn lean_rejects_websocket_accept() {
1620        use eggress_routing::{RouteActionSpec, Router};
1621
1622        let routing: Arc<dyn eggress_routing::RouteService> =
1623            Arc::new(Router::new(vec![], RouteActionSpec::Direct));
1624        let protocols: Arc<[ProtocolId]> = Arc::from([ProtocolId::WebSocket]);
1625        let (mut client, server) = tokio::io::duplex(1024);
1626        let boxed: eggress_core::BoxStream = Box::new(server);
1627        let config = ConnectionConfig {
1628            routing,
1629            context: ConnectionContext::default(),
1630            handshake_timeout: Duration::from_secs(2),
1631            connect_timeout: Duration::from_secs(5),
1632            protocols,
1633            authentication: accept::InboundAuthentication::None,
1634            metrics: None,
1635            udp: None,
1636            tls_client_config: None,
1637            shadowsocks: None,
1638            shadowsocks_metrics: None,
1639            trojan: None,
1640            fixed_target: None,
1641            local_bind: None,
1642        };
1643        client.write_all(&[0xff]).await.unwrap();
1644
1645        let report = serve_connection(boxed, config).await;
1646        assert!(matches!(
1647            report.outcome,
1648            execute::SessionOutcome::ClientProtocolError
1649        ));
1650    }
1651
1652    #[tokio::test]
1653    async fn lean_serves_http_and_socks_normally() {
1654        let (echo_addr, echo_jh) = eggress_testkit::start_echo_server().await;
1655        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1656        let proxy_addr = proxy_listener.local_addr().unwrap();
1657
1658        let proxy_jh = tokio::spawn(async move {
1659            let (stream, _) = proxy_listener.accept().await.unwrap();
1660            let boxed: eggress_core::BoxStream = Box::new(stream);
1661            let routing: Arc<dyn eggress_routing::RouteService> = Arc::new(
1662                eggress_routing::Router::new(vec![], eggress_routing::RouteActionSpec::Direct),
1663            );
1664            let config = ConnectionConfig {
1665                routing,
1666                context: ConnectionContext::default(),
1667                handshake_timeout: Duration::from_secs(5),
1668                connect_timeout: Duration::from_secs(10),
1669                protocols: Arc::from([ProtocolId::Http, ProtocolId::Socks4, ProtocolId::Socks5]),
1670                authentication: accept::InboundAuthentication::None,
1671                metrics: None,
1672                udp: None,
1673                tls_client_config: None,
1674                shadowsocks: None,
1675                shadowsocks_metrics: None,
1676                trojan: None,
1677                fixed_target: None,
1678                local_bind: None,
1679            };
1680            serve_connection(boxed, config).await
1681        });
1682
1683        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
1684        stream.write_all(&[0x05, 0x01, 0x00]).await.unwrap();
1685        let mut response = [0u8; 2];
1686        stream.read_exact(&mut response).await.unwrap();
1687        assert_eq!(response, [0x05, 0x00]);
1688
1689        stream.write_all(&[0x05, 0x01, 0x00, 0x01]).await.unwrap();
1690        match echo_addr.ip() {
1691            std::net::IpAddr::V4(ip) => {
1692                stream.write_all(&ip.octets()).await.unwrap();
1693            }
1694            std::net::IpAddr::V6(ip) => {
1695                stream.write_all(&ip.octets()).await.unwrap();
1696            }
1697        }
1698        stream
1699            .write_all(&echo_addr.port().to_be_bytes())
1700            .await
1701            .unwrap();
1702
1703        let mut reply = [0u8; 10];
1704        stream.read_exact(&mut reply).await.unwrap();
1705        assert_eq!(reply[0], 0x05);
1706        assert_eq!(reply[1], 0x00);
1707
1708        stream.write_all(b"hello").await.unwrap();
1709        stream.shutdown().await.unwrap();
1710
1711        let mut buf = Vec::new();
1712        stream.read_to_end(&mut buf).await.unwrap();
1713        assert_eq!(&buf, b"hello");
1714
1715        let report = proxy_jh.await.unwrap();
1716        assert!(matches!(report.outcome, execute::SessionOutcome::Completed));
1717
1718        echo_jh.abort();
1719    }
1720}
eggress_server/lib.rs

eggress_server/
lib.rs
