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

eggress_server/
lib.rs
