eggress_server/execute/
mod.rs

1//! Session execution: routing, tunnels, HTTP-forward, UDP-associate.
2//!
3//! `execute` turns an `AcceptedSession` into a `SessionReport`;
4//! per-protocol upstream hop handlers live in `hops`.
5
6use crate::accept::{AcceptedSession, PendingHttpForward, PendingTunnel, PendingUdpAssociate};
7use crate::error::SessionOpenError;
8use crate::reply;
9use crate::ConnectionConfig;
10use eggress_core::chain::{ChainExecutor, HopHandler};
11use eggress_core::connector::DirectConnector;
12use eggress_core::relay::relay;
13use eggress_core::BoxStream;
14use eggress_core::{TargetAddr, TargetHost};
15use eggress_routing::{RouteRequest, SelectedRoute};
16use tokio::io::{AsyncReadExt, AsyncWriteExt};
17
18pub(crate) mod hops;
19#[cfg(test)]
20mod tests;
21
22use hops::target_to_socks_addr;
23#[cfg(feature = "pproxy-legacy")]
24use hops::ShadowsocksRHopHandler;
25#[cfg(feature = "ssh")]
26use hops::SshHopHandler;
27use hops::{
28    H2HopHandler, HttpHopHandler, HttpOnlyHopHandler, RawHopHandler, Socks4HopHandler,
29    Socks5HopHandler, UnixHopHandler,
30};
31#[cfg(feature = "quic")]
32use hops::{H3HopHandler, QuicHopHandler};
33#[cfg(feature = "extended")]
34use hops::{ShadowsocksHopHandler, TrojanHopHandler, WebSocketHopHandler};
35
36pub struct SessionReport {
37    pub protocol: Option<String>,
38    pub target: Option<String>,
39    pub route: String,
40    pub bytes_upstream: u64,
41    pub bytes_downstream: u64,
42    pub outcome: SessionOutcome,
43    pub failure: Option<FailureCategory>,
44    pub rule_id: Option<String>,
45    pub upstream_group: Option<String>,
46    pub upstream_id: Option<String>,
47    pub selection_reason: Option<eggress_routing::SelectionReason>,
48}
49
50/// Outcome of a session.
51#[derive(Debug)]
52pub enum SessionOutcome {
53    Completed,
54    ClientProtocolError,
55    AuthenticationFailed,
56    HandshakeTimedOut,
57    RouteFailed,
58    RelayFailed,
59    Cancelled,
60}
61
62/// Specific failure category for structured diagnostics and metrics.
63#[derive(Debug, Clone, Copy, PartialEq, Eq)]
64pub enum FailureCategory {
65    Protocol,
66    Authentication,
67    HandshakeTimeout,
68    Dns,
69    ConnectionRefused,
70    NetworkUnreachable,
71    HostUnreachable,
72    RouteTimeout,
73    RouteHop,
74    UpstreamAuthentication,
75    PolicyDenied,
76    UpstreamUnavailable,
77    Relay,
78    Cancelled,
79    Internal,
80}
81
82impl SessionReport {
83    pub fn open_failed(
84        error: SessionOpenError,
85        protocol: Option<String>,
86        target: Option<String>,
87        route: String,
88    ) -> Self {
89        SessionReport {
90            protocol,
91            target,
92            route,
93            bytes_upstream: 0,
94            bytes_downstream: 0,
95            outcome: SessionOutcome::RouteFailed,
96            failure: Some(FailureCategory::from(&error)),
97            rule_id: None,
98            upstream_group: None,
99            upstream_id: None,
100            selection_reason: None,
101        }
102    }
103
104    pub fn completed(
105        protocol: Option<String>,
106        target: Option<String>,
107        route: String,
108        bytes_upstream: u64,
109        bytes_downstream: u64,
110    ) -> Self {
111        SessionReport {
112            protocol,
113            target,
114            route,
115            bytes_upstream,
116            bytes_downstream,
117            outcome: SessionOutcome::Completed,
118            failure: None,
119            rule_id: None,
120            upstream_group: None,
121            upstream_id: None,
122            selection_reason: None,
123        }
124    }
125
126    pub fn cancelled(protocol: Option<String>, target: Option<String>, route: String) -> Self {
127        SessionReport {
128            protocol,
129            target,
130            route,
131            bytes_upstream: 0,
132            bytes_downstream: 0,
133            outcome: SessionOutcome::Cancelled,
134            failure: Some(FailureCategory::Cancelled),
135            rule_id: None,
136            upstream_group: None,
137            upstream_id: None,
138            selection_reason: None,
139        }
140    }
141
142    pub fn rejected(protocol: Option<String>, target: Option<String>, rule_id: String) -> Self {
143        SessionReport {
144            protocol,
145            target,
146            route: "reject".to_string(),
147            bytes_upstream: 0,
148            bytes_downstream: 0,
149            outcome: SessionOutcome::RouteFailed,
150            failure: Some(FailureCategory::PolicyDenied),
151            rule_id: Some(rule_id),
152            upstream_group: None,
153            upstream_id: None,
154            selection_reason: None,
155        }
156    }
157}
158
159impl From<&SessionOpenError> for FailureCategory {
160    fn from(error: &SessionOpenError) -> Self {
161        match error {
162            SessionOpenError::Dns => FailureCategory::Dns,
163            SessionOpenError::Refused => FailureCategory::ConnectionRefused,
164            SessionOpenError::NetworkUnreachable => FailureCategory::NetworkUnreachable,
165            SessionOpenError::HostUnreachable => FailureCategory::HostUnreachable,
166            SessionOpenError::Timeout => FailureCategory::RouteTimeout,
167            SessionOpenError::UpstreamAuthentication => FailureCategory::UpstreamAuthentication,
168            SessionOpenError::Hop { .. } => FailureCategory::RouteHop,
169            SessionOpenError::PolicyDenied => FailureCategory::PolicyDenied,
170            SessionOpenError::UpstreamUnavailable => FailureCategory::UpstreamUnavailable,
171            SessionOpenError::Other(_) => FailureCategory::Relay,
172        }
173    }
174}
175
176impl FailureCategory {
177    pub fn from_io_error(error: &std::io::Error) -> Self {
178        match error.kind() {
179            std::io::ErrorKind::ConnectionRefused => FailureCategory::ConnectionRefused,
180            std::io::ErrorKind::ConnectionReset => FailureCategory::Relay,
181            std::io::ErrorKind::TimedOut => FailureCategory::Relay,
182            _ => FailureCategory::Relay,
183        }
184    }
185}
186
187/// Execute a session from an accepted connection.
188pub async fn execute(session: AcceptedSession, config: &ConnectionConfig) -> SessionReport {
189    match session {
190        AcceptedSession::Tunnel(pending) => {
191            let protocol = Some(match pending.protocol {
192                crate::accept::TunnelProtocol::HttpConnect => "http".to_string(),
193                crate::accept::TunnelProtocol::Http2 => "h2".to_string(),
194                crate::accept::TunnelProtocol::Http3 => "h3".to_string(),
195                crate::accept::TunnelProtocol::WebSocket => "websocket".to_string(),
196                crate::accept::TunnelProtocol::Socks4 => "socks4".to_string(),
197                crate::accept::TunnelProtocol::Socks5 => "socks5".to_string(),
198                crate::accept::TunnelProtocol::Shadowsocks => "shadowsocks".to_string(),
199                crate::accept::TunnelProtocol::ShadowsocksR => "ssr".to_string(),
200                crate::accept::TunnelProtocol::Trojan => "trojan".to_string(),
201                crate::accept::TunnelProtocol::Raw => "raw".to_string(),
202            });
203            let target = Some(pending.target.to_string());
204            execute_tunnel(pending, config, protocol, target).await
205        }
206        AcceptedSession::HttpForward(pending) => {
207            let target = Some(pending.target.to_string());
208            execute_http_forward(pending, config, target).await
209        }
210        AcceptedSession::UdpAssociate(pending) => execute_udp_associate(pending, config).await,
211        AcceptedSession::Echo(stream) => execute_echo(stream).await,
212    }
213}
214
215async fn execute_echo(mut stream: BoxStream) -> SessionReport {
216    let mut buf = [0u8; 16 * 1024];
217    let mut bytes = 0u64;
218    loop {
219        match stream.read(&mut buf).await {
220            Ok(0) => break,
221            Ok(n) => {
222                bytes += n as u64;
223                if stream.write_all(&buf[..n]).await.is_err() {
224                    break;
225                }
226            }
227            Err(_) => break,
228        }
229    }
230    SessionReport::completed(
231        Some("echo".to_string()),
232        None,
233        "echo".to_string(),
234        bytes,
235        bytes,
236    )
237}
238
239fn route_description(selected: &SelectedRoute) -> String {
240    match selected {
241        SelectedRoute::Direct {
242            selection_reason, ..
243        } => match selection_reason {
244            eggress_routing::SelectionReason::DirectFallback => "direct(fallback)".to_string(),
245            _ => "direct".to_string(),
246        },
247        SelectedRoute::Upstream {
248            upstream, group, ..
249        } => format!("upstream({}/{})", group.0, upstream),
250    }
251}
252
253fn route_metadata(
254    selected: &SelectedRoute,
255) -> (
256    Option<String>,
257    Option<String>,
258    Option<String>,
259    Option<eggress_routing::SelectionReason>,
260) {
261    match selected {
262        SelectedRoute::Direct {
263            decision,
264            selection_reason,
265        } => {
266            let rule_id = match decision {
267                eggress_routing::RouteDecision::Direct { rule, .. }
268                | eggress_routing::RouteDecision::UpstreamGroup { rule, .. }
269                | eggress_routing::RouteDecision::Reject { rule, .. } => rule.0.to_string(),
270            };
271            (Some(rule_id), None, None, Some(*selection_reason))
272        }
273        SelectedRoute::Upstream {
274            decision,
275            group,
276            upstream,
277            selection_reason,
278            ..
279        } => {
280            let rule_id = match decision {
281                eggress_routing::RouteDecision::Direct { rule, .. }
282                | eggress_routing::RouteDecision::UpstreamGroup { rule, .. }
283                | eggress_routing::RouteDecision::Reject { rule, .. } => rule.0.to_string(),
284            };
285            (
286                Some(rule_id),
287                Some(group.0.to_string()),
288                Some(upstream.to_string()),
289                Some(*selection_reason),
290            )
291        }
292    }
293}
294
295struct OpenedRoute {
296    stream: BoxStream,
297    active_lease: Option<eggress_routing::lease::ActiveLease>,
298    route_description: String,
299    rule_id: Option<String>,
300    upstream_group: Option<String>,
301    upstream_id: Option<String>,
302    selection_reason: Option<eggress_routing::SelectionReason>,
303}
304
305fn upstream_protocol_label(chain: &eggress_uri::ProxyChainSpec) -> &'static str {
306    chain
307        .hops
308        .first()
309        .and_then(|h| h.protocols.first())
310        .map(|p| match p {
311            eggress_uri::ProtocolSpec::Http => "http",
312            eggress_uri::ProtocolSpec::HttpOnly => "httponly",
313            eggress_uri::ProtocolSpec::Socks4 => "socks4",
314            eggress_uri::ProtocolSpec::Socks5 => "socks5",
315            eggress_uri::ProtocolSpec::Shadowsocks => "shadowsocks",
316            eggress_uri::ProtocolSpec::ShadowsocksR => "ssr",
317            eggress_uri::ProtocolSpec::Trojan => "trojan",
318            eggress_uri::ProtocolSpec::Http2 => "h2",
319            eggress_uri::ProtocolSpec::Http3 => "h3",
320            eggress_uri::ProtocolSpec::Quic => "quic",
321            eggress_uri::ProtocolSpec::WebSocket => "websocket",
322            eggress_uri::ProtocolSpec::Raw => "raw",
323            eggress_uri::ProtocolSpec::Ssh => "ssh",
324            eggress_uri::ProtocolSpec::Unix => "unix",
325        })
326        .unwrap_or("unknown")
327}
328
329fn failure_reason_label(error: &SessionOpenError) -> &'static str {
330    match error {
331        SessionOpenError::Dns => "dns",
332        SessionOpenError::Refused => "connection_refused",
333        SessionOpenError::NetworkUnreachable => "network_unreachable",
334        SessionOpenError::HostUnreachable => "host_unreachable",
335        SessionOpenError::Timeout => "timeout",
336        SessionOpenError::UpstreamAuthentication => "auth_failed",
337        SessionOpenError::PolicyDenied => "policy_denied",
338        SessionOpenError::Hop { .. } => "handshake",
339        SessionOpenError::UpstreamUnavailable => "upstream_unavailable",
340        SessionOpenError::Other(_) => "io",
341    }
342}
343
344async fn open_route(
345    config: &ConnectionConfig,
346    request: &RouteRequest<'_>,
347) -> Result<OpenedRoute, SessionOpenError> {
348    let selected = config.routing.route(request).map_err(|e| match e {
349        eggress_routing::RouteError::Rejected { .. } => SessionOpenError::PolicyDenied,
350        eggress_routing::RouteError::NoEligibleUpstream(_) => SessionOpenError::PolicyDenied,
351        eggress_routing::RouteError::UnknownGroup(_) => SessionOpenError::PolicyDenied,
352    })?;
353
354    let route = route_description(&selected);
355    let (rule_id, upstream_group, upstream_id, selection_reason) = route_metadata(&selected);
356
357    if let Some(metrics) = &config.metrics {
358        let rule_str = rule_id.as_deref().unwrap_or("default");
359        let action_str = match &selected {
360            SelectedRoute::Direct { .. } => "direct",
361            SelectedRoute::Upstream { .. } => "upstream",
362        };
363        metrics.record_route_decision(rule_str, action_str, "selected");
364    }
365
366    let upstream_protocol = match &selected {
367        SelectedRoute::Upstream { chain, .. } => Some(upstream_protocol_label(chain)),
368        SelectedRoute::Direct { .. } => None,
369    };
370
371    let tls_override = config.tls_client_config.as_ref();
372
373    let result = tokio::time::timeout(config.connect_timeout, async {
374        match selected {
375            SelectedRoute::Direct { .. } => {
376                let bind = config
377                    .local_bind
378                    .as_deref()
379                    .map(|v| {
380                        v.parse().map_err(|e| {
381                            SessionOpenError::Other(format!("invalid local bind '{}': {}", v, e))
382                        })
383                    })
384                    .transpose()?;
385                let stream = DirectConnector
386                    .connect_with_options(
387                        request.target,
388                        &eggress_core::connector::ConnectOptions {
389                            local_bind: bind,
390                            ..Default::default()
391                        },
392                    )
393                    .await?;
394                Ok::<_, SessionOpenError>((stream, None))
395            }
396            SelectedRoute::Upstream {
397                chain,
398                pending_lease,
399                ..
400            } => {
401                #[cfg(feature = "extended")]
402                let shadowsocks_metrics = config.shadowsocks_metrics.clone();
403                #[cfg(not(feature = "extended"))]
404                let shadowsocks_metrics = config.shadowsocks_metrics;
405                #[cfg(feature = "ssh")]
406                let executor = build_chain_executor(
407                    tls_override,
408                    shadowsocks_metrics,
409                    config.ssh_sessions.clone(),
410                );
411                #[cfg(not(feature = "ssh"))]
412                let executor = build_chain_executor(tls_override, shadowsocks_metrics);
413                let stream = executor.execute(&chain.hops, request.target).await?;
414                let active_lease = pending_lease.established();
415                Ok::<_, SessionOpenError>((stream, Some(active_lease)))
416            }
417        }
418    })
419    .await;
420
421    match result {
422        Ok(Ok((stream, active_lease))) => {
423            if let (Some(metrics), Some(protocol)) = (&config.metrics, upstream_protocol) {
424                metrics.record_upstream_open(protocol, "success");
425            }
426            Ok(OpenedRoute {
427                stream,
428                active_lease,
429                route_description: route,
430                rule_id,
431                upstream_group,
432                upstream_id,
433                selection_reason,
434            })
435        }
436        Ok(Err(e)) => {
437            if let (Some(metrics), Some(protocol)) = (&config.metrics, upstream_protocol) {
438                metrics.record_upstream_failure(protocol, failure_reason_label(&e));
439            }
440            Err(e)
441        }
442        Err(_timeout) => {
443            if let Some(metrics) = &config.metrics {
444                if let Some(protocol) = upstream_protocol {
445                    metrics.record_upstream_failure(protocol, "timeout");
446                }
447            }
448            Err(SessionOpenError::Timeout)
449        }
450    }
451}
452
453/// Execute a tunnel session: open route, send success/failure, relay.
454async fn execute_tunnel(
455    mut pending: PendingTunnel,
456    config: &ConnectionConfig,
457    protocol: Option<String>,
458    target: Option<String>,
459) -> SessionReport {
460    tracing::info!("connecting to {}", pending.target);
461
462    let request = RouteRequest {
463        target: &pending.target,
464        source: config.context.source,
465        listener: &config.context.listener,
466        inbound_protocol: match pending.protocol {
467            crate::accept::TunnelProtocol::HttpConnect => eggress_core::ProtocolId::Http,
468            crate::accept::TunnelProtocol::Http2 => eggress_core::ProtocolId::Http2,
469            crate::accept::TunnelProtocol::Http3 => eggress_core::ProtocolId::Http3,
470            crate::accept::TunnelProtocol::WebSocket => eggress_core::ProtocolId::WebSocket,
471            crate::accept::TunnelProtocol::Socks4 => eggress_core::ProtocolId::Socks4,
472            crate::accept::TunnelProtocol::Socks5 => eggress_core::ProtocolId::Socks5,
473            crate::accept::TunnelProtocol::Shadowsocks => eggress_core::ProtocolId::Shadowsocks,
474            crate::accept::TunnelProtocol::ShadowsocksR => eggress_core::ProtocolId::ShadowsocksR,
475            crate::accept::TunnelProtocol::Trojan => eggress_core::ProtocolId::Trojan,
476            crate::accept::TunnelProtocol::Raw => eggress_core::ProtocolId::Raw,
477        },
478        identity: &pending.identity,
479        transport: eggress_routing::TransportKind::Tcp,
480    };
481
482    match open_route(config, &request).await {
483        Ok(opened) => {
484            let route = opened.route_description;
485            let rule_id = opened.rule_id;
486            let upstream_group = opened.upstream_group;
487            let upstream_id = opened.upstream_id;
488            let selection_reason = opened.selection_reason;
489            let _active_lease = opened.active_lease;
490            if let Err(e) = reply::send_tunnel_success(&mut pending, None).await {
491                tracing::debug!("failed to send success reply: {e}");
492                return SessionReport {
493                    protocol,
494                    target,
495                    route,
496                    bytes_upstream: 0,
497                    bytes_downstream: 0,
498                    outcome: SessionOutcome::ClientProtocolError,
499                    failure: Some(FailureCategory::Protocol),
500                    rule_id,
501                    upstream_group,
502                    upstream_id,
503                    selection_reason,
504                };
505            }
506            let result = relay(pending.client, opened.stream).await;
507            tracing::debug!(
508                "relay complete: upstream={}B downstream={}B reason={:?}",
509                result.bytes_upstream,
510                result.bytes_downstream,
511                result.termination_reason
512            );
513            match result.termination_reason {
514                eggress_core::relay::TerminationReason::Error => SessionReport {
515                    protocol,
516                    target,
517                    route,
518                    bytes_upstream: result.bytes_upstream,
519                    bytes_downstream: result.bytes_downstream,
520                    outcome: SessionOutcome::RelayFailed,
521                    failure: Some(FailureCategory::Relay),
522                    rule_id,
523                    upstream_group,
524                    upstream_id,
525                    selection_reason,
526                },
527                _ => SessionReport {
528                    protocol,
529                    target,
530                    route,
531                    bytes_upstream: result.bytes_upstream,
532                    bytes_downstream: result.bytes_downstream,
533                    outcome: SessionOutcome::Completed,
534                    failure: None,
535                    rule_id,
536                    upstream_group,
537                    upstream_id,
538                    selection_reason,
539                },
540            }
541        }
542        Err(SessionOpenError::PolicyDenied) => {
543            let _ = reply::send_tunnel_failure(&mut pending, &SessionOpenError::PolicyDenied).await;
544            SessionReport::rejected(protocol, target, "reject".to_string())
545        }
546        Err(error) => {
547            let _ = reply::send_tunnel_failure(&mut pending, &error).await;
548            SessionReport::open_failed(error, protocol, target, "error".to_string())
549        }
550    }
551}
552
553/// Execute an HTTP forward-proxy session with persistent connection support.
554///
555/// Loops over requests on the client connection, forwarding each to the
556/// appropriate upstream. Supports HTTP/1.1 keep-alive semantics: the
557/// connection persists until the client sends `Connection: close` or the
558/// upstream signals close.
559async fn execute_http_forward(
560    pending: PendingHttpForward,
561    config: &ConnectionConfig,
562    _target: Option<String>,
563) -> SessionReport {
564    tracing::info!("forward proxy to {}", pending.target);
565
566    let mut client = pending.client;
567    let mut total_bytes_upstream: u64 = 0;
568    let mut total_bytes_downstream: u64 = 0;
569    let mut last_target: Option<String>;
570    let mut last_rule_id: Option<String> = None;
571    let mut last_upstream_group: Option<String> = None;
572    let mut last_upstream_id: Option<String> = None;
573    let mut last_selection_reason: Option<eggress_routing::SelectionReason> = None;
574    let mut last_route = String::new();
575
576    // Process the first request (already parsed in pending)
577    let mut request = pending.request;
578    let mut client_close = request.connection_close;
579
580    loop {
581        let target_addr = request.target.clone();
582        last_target = Some(target_addr.to_string());
583
584        if eggress_protocol_http::has_unsupported_expectation(&request.headers) {
585            let _ = reply::send_http_expectation_failed(&mut client).await;
586            return SessionReport {
587                protocol: None,
588                target: last_target,
589                route: last_route,
590                bytes_upstream: total_bytes_upstream,
591                bytes_downstream: total_bytes_downstream,
592                outcome: SessionOutcome::ClientProtocolError,
593                failure: Some(FailureCategory::Protocol),
594                rule_id: last_rule_id,
595                upstream_group: last_upstream_group,
596                upstream_id: last_upstream_id,
597                selection_reason: last_selection_reason,
598            };
599        }
600
601        let route_request = RouteRequest {
602            target: &target_addr,
603            source: config.context.source,
604            listener: &config.context.listener,
605            inbound_protocol: eggress_core::ProtocolId::Http,
606            identity: &pending.identity,
607            transport: eggress_routing::TransportKind::Tcp,
608        };
609
610        match open_route(config, &route_request).await {
611            Ok(mut opened) => {
612                last_route = opened.route_description;
613                last_rule_id = opened.rule_id;
614                last_upstream_group = opened.upstream_group;
615                last_upstream_id = opened.upstream_id;
616                last_selection_reason = opened.selection_reason;
617                let _active_lease = opened.active_lease;
618
619                let origin_req = eggress_protocol_http::build_origin_request(&request);
620                let head_bytes = origin_req.len() as u64;
621
622                if let Err(e) = opened.stream.write_all(origin_req.as_bytes()).await {
623                    let _ = reply::send_http_forward_failure(
624                        &mut client,
625                        &SessionOpenError::Other(e.to_string()),
626                    )
627                    .await;
628                    return SessionReport {
629                        protocol: None,
630                        target: last_target,
631                        route: last_route,
632                        bytes_upstream: total_bytes_upstream,
633                        bytes_downstream: total_bytes_downstream,
634                        outcome: SessionOutcome::RelayFailed,
635                        failure: Some(FailureCategory::Relay),
636                        rule_id: last_rule_id,
637                        upstream_group: last_upstream_group,
638                        upstream_id: last_upstream_id,
639                        selection_reason: last_selection_reason,
640                    };
641                }
642                if let Err(e) = opened.stream.flush().await {
643                    let _ = reply::send_http_forward_failure(
644                        &mut client,
645                        &SessionOpenError::Other(e.to_string()),
646                    )
647                    .await;
648                    return SessionReport {
649                        protocol: None,
650                        target: last_target,
651                        route: last_route,
652                        bytes_upstream: total_bytes_upstream + head_bytes,
653                        bytes_downstream: total_bytes_downstream,
654                        outcome: SessionOutcome::RelayFailed,
655                        failure: Some(FailureCategory::Relay),
656                        rule_id: last_rule_id,
657                        upstream_group: last_upstream_group,
658                        upstream_id: last_upstream_id,
659                        selection_reason: last_selection_reason,
660                    };
661                }
662
663                // The connect timeout ends once the upstream route has been
664                // opened. Uploading a request body is an independent stream
665                // operation and may legitimately take longer.
666                let body_result = async {
667                    let report = eggress_protocol_http::copy_request_body(
668                        &mut client,
669                        &mut opened.stream,
670                        request.body_kind(),
671                        &eggress_protocol_http::BodyCopyLimits::default(),
672                    )
673                    .await?;
674                    opened.stream.flush().await?;
675                    Ok::<_, eggress_protocol_http::HttpError>(report)
676                }
677                .await;
678                let body_report = match body_result {
679                    Ok(report) => report,
680                    Err(_) => {
681                        let _ = opened.stream.shutdown().await;
682                        let _ = client.shutdown().await;
683                        return SessionReport {
684                            protocol: None,
685                            target: last_target,
686                            route: last_route,
687                            bytes_upstream: total_bytes_upstream + head_bytes,
688                            bytes_downstream: total_bytes_downstream,
689                            outcome: SessionOutcome::ClientProtocolError,
690                            failure: Some(FailureCategory::Protocol),
691                            rule_id: last_rule_id,
692                            upstream_group: last_upstream_group,
693                            upstream_id: last_upstream_id,
694                            selection_reason: last_selection_reason,
695                        };
696                    }
697                };
698
699                total_bytes_upstream += head_bytes + body_report.wire_bytes;
700
701                let forward_result =
702                    match eggress_protocol_http::forward_response(&mut opened.stream, &mut client)
703                        .await
704                    {
705                        Ok(result) => result,
706                        Err(eggress_protocol_http::HttpError::UpgradeUnsupported) => {
707                            let _ = reply::send_http_upgrade_unsupported(&mut client).await;
708                            return SessionReport {
709                                protocol: None,
710                                target: last_target,
711                                route: last_route,
712                                bytes_upstream: total_bytes_upstream,
713                                bytes_downstream: total_bytes_downstream,
714                                outcome: SessionOutcome::RelayFailed,
715                                failure: Some(FailureCategory::Protocol),
716                                rule_id: last_rule_id,
717                                upstream_group: last_upstream_group,
718                                upstream_id: last_upstream_id,
719                                selection_reason: last_selection_reason,
720                            };
721                        }
722                        Err(_e) => {
723                            let _ = client.shutdown().await;
724                            return SessionReport {
725                                protocol: None,
726                                target: last_target,
727                                route: last_route,
728                                bytes_upstream: total_bytes_upstream,
729                                bytes_downstream: total_bytes_downstream,
730                                outcome: SessionOutcome::RelayFailed,
731                                failure: Some(FailureCategory::Relay),
732                                rule_id: last_rule_id,
733                                upstream_group: last_upstream_group,
734                                upstream_id: last_upstream_id,
735                                selection_reason: last_selection_reason,
736                            };
737                        }
738                    };
739
740                total_bytes_downstream += forward_result.report.bytes_forwarded;
741
742                // Determine whether to continue the session
743                let should_close = client_close
744                    || forward_result.client_should_close
745                    || !forward_result.upstream_alive;
746
747                if should_close {
748                    break;
749                }
750
751                // Read the next request from the client
752                match eggress_protocol_http::forward_request_stream(&mut client).await {
753                    Ok(next_request) => {
754                        client_close = next_request.connection_close;
755                        request = next_request;
756                    }
757                    Err(eggress_protocol_http::HttpError::Io(ref e))
758                        if e.kind() == std::io::ErrorKind::UnexpectedEof =>
759                    {
760                        // Client closed the connection
761                        break;
762                    }
763                    Err(_) => {
764                        // Malformed next request — close with error
765                        break;
766                    }
767                }
768            }
769            Err(SessionOpenError::PolicyDenied) => {
770                let _ =
771                    reply::send_http_forward_failure(&mut client, &SessionOpenError::PolicyDenied)
772                        .await;
773                return SessionReport::rejected(None, last_target, "reject".to_string());
774            }
775            Err(error) => {
776                let _ = reply::send_http_forward_failure(&mut client, &error).await;
777                return SessionReport::open_failed(error, None, last_target, "error".to_string());
778            }
779        }
780    }
781
782    SessionReport {
783        protocol: None,
784        target: last_target,
785        route: last_route,
786        bytes_upstream: total_bytes_upstream,
787        bytes_downstream: total_bytes_downstream,
788        outcome: SessionOutcome::Completed,
789        failure: None,
790        rule_id: last_rule_id,
791        upstream_group: last_upstream_group,
792        upstream_id: last_upstream_id,
793        selection_reason: last_selection_reason,
794    }
795}
796
797async fn execute_udp_associate(
798    pending: PendingUdpAssociate,
799    config: &ConnectionConfig,
800) -> SessionReport {
801    let protocol = Some("socks5".to_string());
802
803    let udp_service = match &config.udp {
804        Some(svc) if svc.is_enabled() => svc,
805        _ => {
806            tracing::debug!("UDP ASSOCIATE rejected: UDP service not available");
807            let mut stream = pending.client;
808            let target = pending.client_hint.unwrap_or(TargetAddr {
809                host: TargetHost::Ip(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED)),
810                port: 0,
811            });
812            let socks_addr = target_to_socks_addr(&target);
813            let _ = eggress_protocol_socks::socks5::server::send_connect_reply(
814                &mut stream,
815                eggress_protocol_socks::socks5::server::REP_NOT_ALLOWED,
816                &socks_addr,
817            )
818            .await;
819            return SessionReport {
820                protocol,
821                target: None,
822                route: "udp_associate_disabled".to_string(),
823                bytes_upstream: 0,
824                bytes_downstream: 0,
825                outcome: SessionOutcome::RouteFailed,
826                failure: Some(FailureCategory::Protocol),
827                rule_id: None,
828                upstream_group: None,
829                upstream_id: None,
830                selection_reason: None,
831            };
832        }
833    };
834
835    let client_tcp_peer = config.context.source;
836
837    let gen = config.context.generation;
838
839    let handle = match tokio::time::timeout(
840        config.connect_timeout,
841        udp_service.create_association(
842            &config.context.listener,
843            client_tcp_peer,
844            pending.identity.clone(),
845            gen,
846        ),
847    )
848    .await
849    {
850        Ok(Ok(handle)) => handle,
851        Ok(Err(e)) => {
852            tracing::debug!("UDP ASSOCIATE failed: {e}");
853            let mut stream = pending.client;
854            let target = pending.client_hint.unwrap_or(TargetAddr {
855                host: TargetHost::Ip(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED)),
856                port: 0,
857            });
858            let socks_addr = target_to_socks_addr(&target);
859            let _ = eggress_protocol_socks::socks5::server::send_connect_reply(
860                &mut stream,
861                eggress_protocol_socks::socks5::server::REP_GENERAL_FAILURE,
862                &socks_addr,
863            )
864            .await;
865            return SessionReport {
866                protocol,
867                target: None,
868                route: "udp_associate_failed".to_string(),
869                bytes_upstream: 0,
870                bytes_downstream: 0,
871                outcome: SessionOutcome::RouteFailed,
872                failure: Some(FailureCategory::Protocol),
873                rule_id: None,
874                upstream_group: None,
875                upstream_id: None,
876                selection_reason: None,
877            };
878        }
879        Err(_) => {
880            tracing::debug!("UDP ASSOCIATE failed: timeout");
881            let mut stream = pending.client;
882            let target = pending.client_hint.unwrap_or(TargetAddr {
883                host: TargetHost::Ip(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED)),
884                port: 0,
885            });
886            let socks_addr = target_to_socks_addr(&target);
887            let _ = eggress_protocol_socks::socks5::server::send_connect_reply(
888                &mut stream,
889                eggress_protocol_socks::socks5::server::REP_GENERAL_FAILURE,
890                &socks_addr,
891            )
892            .await;
893            return SessionReport {
894                protocol,
895                target: None,
896                route: "udp_associate_timeout".to_string(),
897                bytes_upstream: 0,
898                bytes_downstream: 0,
899                outcome: SessionOutcome::HandshakeTimedOut,
900                failure: Some(FailureCategory::RouteTimeout),
901                rule_id: None,
902                upstream_group: None,
903                upstream_id: None,
904                selection_reason: None,
905            };
906        }
907    };
908
909    let relay_ip = handle.relay_addr.ip();
910    let relay_port = handle.relay_addr.port();
911    let socks_addr = match relay_ip {
912        std::net::IpAddr::V4(ip) => {
913            eggress_protocol_socks::socks5::server::SocksAddr::IPv4(ip.octets(), relay_port)
914        }
915        std::net::IpAddr::V6(ip) => {
916            eggress_protocol_socks::socks5::server::SocksAddr::IPv6(ip.octets(), relay_port)
917        }
918    };
919
920    let mut stream = pending.client;
921    if let Err(e) =
922        eggress_protocol_socks::socks5::server::send_udp_associate_reply(&mut stream, &socks_addr)
923            .await
924    {
925        tracing::debug!("failed to send UDP ASSOCIATE reply: {e}");
926        handle.cancel.cancel();
927        return SessionReport {
928            protocol,
929            target: None,
930            route: "udp_associate_reply_failed".to_string(),
931            bytes_upstream: 0,
932            bytes_downstream: 0,
933            outcome: SessionOutcome::ClientProtocolError,
934            failure: Some(FailureCategory::Protocol),
935            rule_id: None,
936            upstream_group: None,
937            upstream_id: None,
938            selection_reason: None,
939        };
940    }
941
942    tracing::info!(
943        association_id = ?handle.id,
944        relay_addr = %handle.relay_addr,
945        "UDP ASSOCIATE established, keeping TCP control connection alive"
946    );
947
948    let mut buf = [0u8; 1];
949    tokio::select! {
950        result = stream.read_exact(&mut buf) => {
951            match result {
952                Ok(_) => {
953                    tracing::debug!(
954                        association_id = ?handle.id,
955                        "TCP control connection closed by client"
956                    );
957                }
958                Err(_) => {
959                    tracing::debug!(
960                        association_id = ?handle.id,
961                        "TCP control connection read failed"
962                    );
963                }
964            }
965        }
966        _ = handle.cancel.cancelled() => {
967            tracing::debug!(
968                association_id = ?handle.id,
969                "UDP association cancelled"
970            );
971        }
972    }
973
974    handle.cancel.cancel();
975
976    SessionReport {
977        protocol,
978        target: None,
979        route: "udp_associate".to_string(),
980        bytes_upstream: 0,
981        bytes_downstream: 0,
982        outcome: SessionOutcome::Completed,
983        failure: None,
984        rule_id: None,
985        upstream_group: None,
986        upstream_id: None,
987        selection_reason: None,
988    }
989}
990
991pub fn build_chain_executor(
992    tls_override: Option<&std::sync::Arc<rustls::ClientConfig>>,
993    #[cfg(feature = "extended")] shadowsocks_metrics: Option<
994        std::sync::Arc<eggress_protocol_shadowsocks::ShadowsocksMetrics>,
995    >,
996    #[cfg(not(feature = "extended"))] _shadowsocks_metrics: Option<()>,
997    #[cfg(feature = "ssh")] ssh_sessions: Option<
998        std::sync::Arc<eggress_transport_ssh::SshSessionCache>,
999    >,
1000) -> ChainExecutor {
1001    // Build shared TLS client config for upstream hops
1002    let shared_tls_config = match tls_override {
1003        Some(config) => Some(config.clone()),
1004        None => {
1005            let builder = eggress_transport_tls::TlsClientConfigBuilder::new();
1006            match builder.with_system_roots().and_then(|b| b.build()) {
1007                Ok(config) => Some(config),
1008                Err(e) => {
1009                    tracing::warn!("failed to build shared TLS config: {e}");
1010                    None
1011                }
1012            }
1013        }
1014    };
1015
1016    #[cfg(feature = "extended")]
1017    let shared_tls_config_arc = shared_tls_config.clone();
1018    #[cfg(not(feature = "extended"))]
1019    let _shared_tls_config_arc = shared_tls_config.clone();
1020
1021    // Per-hop `?insecure` requires an insecure verifier. Build it only when
1022    // the `insecure-tls` feature is available; otherwise per-hop insecure hops
1023    // will be rejected in `ChainExecutor::validate_chain` / `execute` with an
1024    // explicit error. The transport's `with_insecure` is feature-gated, so
1025    // `cargo test` without the feature intentionally leaves this as `None`.
1026    #[cfg(feature = "insecure-tls")]
1027    let insecure_shared_tls_config: Option<std::sync::Arc<rustls::ClientConfig>> =
1028        if tls_override.is_some() {
1029            None
1030        } else {
1031            let builder = eggress_transport_tls::TlsClientConfigBuilder::new();
1032            match builder
1033                .with_system_roots()
1034                .map(|b| b.with_insecure())
1035                .and_then(|b| b.build())
1036            {
1037                Ok(cfg) => Some(cfg),
1038                Err(e) => {
1039                    tracing::debug!("failed to build insecure TLS config: {e}");
1040                    None
1041                }
1042            }
1043        };
1044    #[cfg(not(feature = "insecure-tls"))]
1045    let insecure_shared_tls_config: Option<std::sync::Arc<rustls::ClientConfig>> = None;
1046
1047    let mut handlers: Vec<Box<dyn HopHandler>> = vec![
1048        Box::new(HttpHopHandler),
1049        Box::new(HttpOnlyHopHandler),
1050        Box::new(Socks5HopHandler),
1051        Box::new(Socks4HopHandler),
1052    ];
1053
1054    #[cfg(feature = "extended")]
1055    {
1056        handlers.push(Box::new(ShadowsocksHopHandler {
1057            metrics: shadowsocks_metrics,
1058        }));
1059        handlers.push(Box::new(TrojanHopHandler {
1060            tls_config: shared_tls_config_arc.clone(),
1061            insecure_tls_config: insecure_shared_tls_config.clone(),
1062            tls_override: tls_override.cloned(),
1063        }));
1064        handlers.push(Box::new(WebSocketHopHandler));
1065    }
1066
1067    #[cfg(feature = "pproxy-legacy")]
1068    handlers.push(Box::new(ShadowsocksRHopHandler));
1069
1070    handlers.push(Box::new(RawHopHandler));
1071    handlers.push(Box::new(UnixHopHandler));
1072    #[cfg(feature = "ssh")]
1073    if let Some(sessions) = ssh_sessions {
1074        handlers.push(Box::new(SshHopHandler { sessions }));
1075    }
1076    handlers.push(Box::new(H2HopHandler));
1077
1078    #[cfg(feature = "quic")]
1079    {
1080        handlers.push(Box::new(QuicHopHandler));
1081        handlers.push(Box::new(H3HopHandler));
1082    }
1083
1084    // Pre-build TLS configs per distinct ALPN set so we don't re-read
1085    // and re-parse system roots on every handshake (O-05).
1086    let tls_wrapper_default = shared_tls_config.clone();
1087    let tls_wrapper_h2: Option<std::sync::Arc<rustls::ClientConfig>> = if tls_override.is_none() {
1088        let builder = eggress_transport_tls::TlsClientConfigBuilder::new();
1089        match builder.with_system_roots().and_then(|b| {
1090            b.with_alpn(vec![b"h2".to_vec(), b"http/1.1".to_vec()])
1091                .build()
1092        }) {
1093            Ok(cfg) => Some(cfg),
1094            Err(e) => {
1095                tracing::debug!("failed to build h2 TLS config: {e}");
1096                None
1097            }
1098        }
1099    } else {
1100        None
1101    };
1102    #[cfg(feature = "insecure-tls")]
1103    let insecure_wrapper_default = insecure_shared_tls_config.clone();
1104    #[cfg(not(feature = "insecure-tls"))]
1105    let insecure_wrapper_default: Option<std::sync::Arc<rustls::ClientConfig>> = None;
1106    #[cfg(feature = "insecure-tls")]
1107    let insecure_wrapper_h2: Option<std::sync::Arc<rustls::ClientConfig>> =
1108        if tls_override.is_none() && insecure_shared_tls_config.is_some() {
1109            let builder = eggress_transport_tls::TlsClientConfigBuilder::new();
1110            match builder
1111                .with_system_roots()
1112                .map(|b| b.with_insecure())
1113                .and_then(|b| {
1114                    b.with_alpn(vec![b"h2".to_vec(), b"http/1.1".to_vec()])
1115                        .build()
1116                }) {
1117                Ok(cfg) => Some(cfg),
1118                Err(e) => {
1119                    tracing::debug!("failed to build insecure h2 TLS config: {e}");
1120                    None
1121                }
1122            }
1123        } else {
1124            None
1125        };
1126    #[cfg(not(feature = "insecure-tls"))]
1127    let insecure_wrapper_h2: Option<std::sync::Arc<rustls::ClientConfig>> = None;
1128    fn build_alpn_config(
1129        alpn: Option<Vec<Vec<u8>>>,
1130    ) -> Result<std::sync::Arc<rustls::ClientConfig>, Box<dyn std::error::Error + Send + Sync>>
1131    {
1132        let mut builder = eggress_transport_tls::TlsClientConfigBuilder::new();
1133        builder = builder.with_system_roots()?;
1134        if let Some(protocols) = alpn {
1135            builder = builder.with_alpn(protocols);
1136        }
1137        Ok(builder.build()?)
1138    }
1139    #[cfg(feature = "insecure-tls")]
1140    fn build_insecure_alpn_config(
1141        alpn: Option<Vec<Vec<u8>>>,
1142    ) -> Result<std::sync::Arc<rustls::ClientConfig>, Box<dyn std::error::Error + Send + Sync>>
1143    {
1144        let mut builder = eggress_transport_tls::TlsClientConfigBuilder::new();
1145        builder = builder.with_system_roots()?;
1146        builder = builder.with_insecure();
1147        if let Some(protocols) = alpn {
1148            builder = builder.with_alpn(protocols);
1149        }
1150        Ok(builder.build()?)
1151    }
1152    #[cfg(not(feature = "insecure-tls"))]
1153    fn build_insecure_alpn_config(
1154        _alpn: Option<Vec<Vec<u8>>>,
1155    ) -> Result<std::sync::Arc<rustls::ClientConfig>, Box<dyn std::error::Error + Send + Sync>>
1156    {
1157        Err("insecure TLS requires the insecure-tls feature".into())
1158    }
1159    let tls_wrapper: eggress_core::chain::TlsWrapper =
1160        Box::new(move |stream, server_name, alpn, insecure| {
1161            let default = tls_wrapper_default.clone();
1162            let h2_cfg = tls_wrapper_h2.clone();
1163            let insecure_default = insecure_wrapper_default.clone();
1164            let insecure_h2_cfg = insecure_wrapper_h2.clone();
1165            Box::pin(async move {
1166                let config = if insecure {
1167                    match insecure_default.clone() {
1168                        Some(c) => {
1169                            if let Some(ref protocols) = alpn {
1170                                if c.alpn_protocols == *protocols {
1171                                    c
1172                                } else if let Some(h2) = insecure_h2_cfg.clone() {
1173                                    if *protocols == vec![b"h2".to_vec(), b"http/1.1".to_vec()] {
1174                                        h2
1175                                    } else {
1176                                        build_insecure_alpn_config(Some(protocols.clone()))?
1177                                    }
1178                                } else {
1179                                    build_insecure_alpn_config(Some(protocols.clone()))?
1180                                }
1181                            } else {
1182                                c
1183                            }
1184                        }
1185                        None => build_insecure_alpn_config(alpn)?,
1186                    }
1187                } else {
1188                    match default {
1189                        Some(c) => {
1190                            if let Some(ref protocols) = alpn {
1191                                if c.alpn_protocols == *protocols {
1192                                    c
1193                                } else if let Some(h2) = h2_cfg {
1194                                    if *protocols == vec![b"h2".to_vec(), b"http/1.1".to_vec()] {
1195                                        h2
1196                                    } else {
1197                                        build_alpn_config(Some(protocols.clone()))?
1198                                    }
1199                                } else {
1200                                    build_alpn_config(Some(protocols.clone()))?
1201                                }
1202                            } else {
1203                                c
1204                            }
1205                        }
1206                        None => build_alpn_config(alpn)?,
1207                    }
1208                };
1209                eggress_transport_tls::tls_connect(stream, config, &server_name)
1210                    .await
1211                    .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { Box::new(e) as _ })
1212            })
1213        });
1214
1215    ChainExecutor::new(handlers)
1216        .with_tls_wrapper(tls_wrapper)
1217        .with_shared_tls_config(shared_tls_config)
1218        .with_insecure_shared_tls_config(insecure_shared_tls_config)
1219}
eggress_server/execute/mod.rs

eggress_server/execute/
mod.rs
