eggress_server/
lib.rs

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

eggress_server/
lib.rs
