geiserx_ts_runtime 0.57.7

//! Owns the inbound subnet-router / exit-node forwarder and its dedicated any-IP netstack.
//!
//! The forwarder netstack is *separate* from the application netstack ([`NetstackActor`]): it has
//! any-IP acceptance enabled so it captures inbound overlay flows addressed to destinations this
//! node does not own (the advertised subnet routes / exit-node default route), and splices them to
//! real OS sockets via [`ts_forwarder`]. Routing of inbound packets to this netstack's transport is
//! done in [`route_updater`](crate::route_updater): advertised prefixes resolve to this transport,
//! the node's own addresses to the application transport.
//!
//! [`NetstackActor`]: crate::netstack_actor::NetstackActor

use kameo::actor::ActorRef;
use netstack::{
    HasChannel,
    netcore::{Channel, NetstackControl},
};
use tokio::task::JoinSet;
use ts_forwarder::{
    Admit, DirectDialer, ExitDnsDialer, Forwarder, HostExitDialer, PortGate, ProxyExitDialer,
    RealDialer, RouteTable, RouteUpdater,
};

use crate::{
    Error,
    dataplane::{OverlayFromDataplane, OverlayToDataplane},
    env::Env,
};

pub struct ForwarderActor {
    _joinset: JoinSet<()>,
    channel: Channel,
    /// Live handle to swap the forwarder's accept/dial route table at runtime (a
    /// `set_advertise_routes`). New flows see the new set; in-flight flows keep their classification.
    route_updater: RouteUpdater,
}

/// Build a [`Forwarder`] with the given dialer and spawn its run loop onto `joinset`.
///
/// Generic over the concrete [`RealDialer`] so the fail-closed [`DirectDialer`] and the opt-in
/// [`HostExitDialer`] share one run-loop body — only the dialer type differs, so the two gate arms
/// can't drift. When `all_ports` is set the explicit `tcp_ports`/`udp_ports` sets are ignored and
/// the forwarder runs in gated all-port mode, returning the [`PortGate`] every inbound IPv4 packet
/// must be offered to (instead of writing it to `ingress`); otherwise it forwards exactly the
/// configured port sets.
#[allow(clippy::too_many_arguments)]
fn spawn_forwarder<D: RealDialer>(
    joinset: &mut JoinSet<()>,
    channel: Channel,
    routes: RouteTable,
    dialer: D,
    all_ports: bool,
    tcp_ports: Vec<u16>,
    udp_ports: Vec<u16>,
    ingress: netstack::WakingPipeSender,
) -> (RouteUpdater, Option<PortGate>) {
    let (forwarder, gate) = match forwarder_mode(all_ports) {
        ForwarderMode::AllPorts => {
            let (f, g) = Forwarder::all_ports_gated(channel, routes, dialer, ingress);
            (f, Some(g))
        }
        ForwarderMode::Ports => (
            Forwarder::new(channel, routes, dialer, tcp_ports, udp_ports),
            None,
        ),
    };
    // Grab the live route-update handle BEFORE `run()` consumes the forwarder, so the actor can
    // push a new `RouteTable` (a runtime `set_advertise_routes`) onto the running forwarder. New
    // flows pick up the change; in-flight flows keep their classification.
    let route_updater = forwarder.route_updater();
    joinset.spawn(async move {
        if let Err(e) = forwarder.run().await {
            tracing::error!(error = %e, "forwarder run loop exited");
        }
    });
    (route_updater, gate)
}

/// Token bucket for IPv6 rejections (TCP RST / ICMPv6) the forwarder sends back to peers:
/// [`RejectBudget::PER_SEC`] per second, bursting to the same.
struct RejectBudget {
    tokens: f64,
    last: std::time::Instant,
}

impl RejectBudget {
    const PER_SEC: f64 = 1000.0;

    fn new() -> Self {
        Self {
            tokens: Self::PER_SEC,
            last: std::time::Instant::now(),
        }
    }

    fn take(&mut self) -> bool {
        let now = std::time::Instant::now();
        let refill = now.duration_since(self.last).as_secs_f64() * Self::PER_SEC;
        self.tokens = (self.tokens + refill).min(Self::PER_SEC);
        self.last = now;
        if self.tokens >= 1.0 {
            self.tokens -= 1.0;
            true
        } else {
            false
        }
    }
}

/// Which concrete dialer the forwarder is wired with — the anti-leak gate's only output.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DialerChoice {
    /// Fail-closed default: structurally refuses exit-node egress.
    Direct,
    /// Explicit opt-in: egresses exit-node flows via this host's real IP.
    HostExit,
    /// Explicit opt-in: egresses exit-node flows through an upstream proxy (fails closed).
    Proxy,
}

/// Pure selection of the forwarder dialer from the `forward_exit_egress` gate and whether an exit
/// proxy is configured, factored out of `on_start` so it can be unit-tested without a netstack.
///
/// - exit egress off => fail-closed `DirectDialer` (a proxy config is ignored unless egress is on).
/// - exit egress on, proxy configured => `ProxyExitDialer` (egress via the proxy IP, fail-closed).
/// - exit egress on, no proxy => `HostExitDialer` (egress via this host's real IP).
fn dialer_choice(forward_exit_egress: bool, has_exit_proxy: bool) -> DialerChoice {
    match (forward_exit_egress, has_exit_proxy) {
        (false, _) => DialerChoice::Direct,
        (true, true) => DialerChoice::Proxy,
        (true, false) => DialerChoice::HostExit,
    }
}

/// Whether the forwarder runs in all-port mode or forwards an explicit port set.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ForwarderMode {
    /// All TCP/UDP ports per advertised route (gated on-demand listeners).
    AllPorts,
    /// Exactly the configured TCP/UDP port sets.
    Ports,
}

/// Pure selection of the forwarder port mode from the `forward_all_ports` flag. All-port mode is
/// chosen iff (and only iff) `forward_all_ports` is set; otherwise the explicit port sets.
fn forwarder_mode(forward_all_ports: bool) -> ForwarderMode {
    if forward_all_ports {
        ForwarderMode::AllPorts
    } else {
        ForwarderMode::Ports
    }
}

impl kameo::Actor for ForwarderActor {
    type Args = (
        Env,
        netstack::netcore::Config,
        OverlayToDataplane,
        OverlayFromDataplane,
    );
    type Error = Error;

    async fn on_start(
        (env, config, netstack_up, netstack_down): Self::Args,
        _slf: ActorRef<Self>,
    ) -> Result<Self, Self::Error> {
        let (
            mut netstack,
            netstack::WakingPipe {
                rx: mut netstack_down_rx,
                tx: netstack_down_tx,
            },
        ) = crate::netstack_actor::piped_ingress_bounded(config);
        let channel = netstack.command_channel();

        let mut joinset = JoinSet::new();

        joinset.spawn(async move {
            netstack.run_tokio().await;
        });

        // Pump packets emitted by the forwarder netstack down into the dataplane (out the overlay).
        let reject_up = netstack_up.clone();
        joinset.spawn(async move {
            while let Some(first) = netstack_down_rx.recv_async().await {
                // Hand the dataplane whatever is already queued as one batch, so encryption and the
                // underlay send can work per batch rather than per packet.
                let mut batch = Vec::with_capacity(crate::netstack_actor::EGRESS_BATCH);
                batch.push(crate::netstack_actor::into_packet(first));
                while batch.len() < crate::netstack_actor::EGRESS_BATCH
                    && let Some(more) = netstack_down_rx.try_recv()
                {
                    batch.push(crate::netstack_actor::into_packet(more));
                }
                if netstack_up.send(batch).is_err() {
                    break;
                }
            }

            tracing::warn!("forwarder netstack downlink shut down!");
        });

        // Enable any-IP acceptance BEFORE the forwarder starts accepting, so the first inbound flow
        // to a foreign destination is captured rather than rejected. This is the dedicated forwarder
        // netstack; never the application netstack (see `SetAnyIp` safety constraints). The netstack
        // run loop is already spawned above, so this round-trips. A failure here means the freshly
        // spawned netstack channel is already gone — a fatal startup error.
        if let Err(e) = channel.set_any_ip(true).await {
            tracing::error!(error = %e, "enabling any-IP on forwarder netstack");
            return Err(Error {
                kind: crate::ErrorKind::ActorGone,
                message_ty: None,
                target_actor: None,
            });
        }

        // The forwarder dials precisely the prefixes we advertise (advertise == forward). The dialer
        // is the single anti-leak chokepoint, selected here by the `forward_exit_egress` gate plus
        // whether an upstream exit proxy is configured:
        //
        // - `DirectDialer` (default, fail-closed): dials real sockets bound to 0.0.0.0:0 for subnet
        //   routes and *structurally refuses* exit-node egress, so a 0.0.0.0/0 flow routed to this
        //   netstack is dropped at dial time, never leaked out our real IP.
        // - `HostExitDialer` (explicit opt-in, no proxy configured): also egresses exit-node flows
        //   via this host's real IP. Chosen only when the operator set `forward_exit_egress`, which
        //   is an auditable, deliberate act (see its config docs).
        // - `ProxyExitDialer` (explicit opt-in, exit proxy configured): egresses exit-node flows
        //   through the configured upstream proxy (e.g. a residential proxy), so the node's real
        //   origin IP never leaves. Fails closed — any proxy connect/handshake failure drops the
        //   flow rather than falling back to a direct host-IP dial.
        //
        // The dialers are distinct concrete types (`Forwarder<D>` is generic), so we branch on the
        // gate to pick the dialer but funnel all arms through one `spawn_forwarder` helper — the
        // run-loop body lives in exactly one place so the fail-closed and opt-in arms can't drift.
        let routes = RouteTable::new(env.forward_routes.iter().copied());
        let all_ports = env.forward_all_ports;
        let tcp_ports = env.forward_tcp_ports.as_ref().clone();
        let udp_ports = env.forward_udp_ports.as_ref().clone();
        let n_routes = env.forward_routes.len();
        let n_tcp_ports = tcp_ports.len();
        let n_udp_ports = udp_ports.len();

        let choice = dialer_choice(env.forward_exit_egress, env.exit_proxy.is_some());
        let (route_updater, gate) = match choice {
            DialerChoice::Proxy => {
                // `dialer_choice` returns `Proxy` only when `exit_proxy.is_some()`, so this clone
                // is always present; expressed as an expect so a future gate change can't silently
                // fall through to a direct dial (which would leak the real IP).
                let proxy_config = env
                    .exit_proxy
                    .clone()
                    .expect("dialer_choice returned Proxy without an exit proxy configured");
                spawn_forwarder(
                    &mut joinset,
                    channel.clone(),
                    routes,
                    ProxyExitDialer::new(proxy_config),
                    all_ports,
                    tcp_ports,
                    udp_ports,
                    netstack_down_tx.clone(),
                )
            }
            DialerChoice::HostExit if env.exit_node_dns.is_some() => spawn_forwarder(
                &mut joinset,
                channel.clone(),
                routes,
                ExitDnsDialer::new(HostExitDialer, env.exit_node_dns.expect("checked by guard")),
                all_ports,
                tcp_ports,
                udp_ports,
                netstack_down_tx.clone(),
            ),
            DialerChoice::HostExit => spawn_forwarder(
                &mut joinset,
                channel.clone(),
                routes,
                HostExitDialer,
                all_ports,
                tcp_ports,
                udp_ports,
                netstack_down_tx.clone(),
            ),
            DialerChoice::Direct => spawn_forwarder(
                &mut joinset,
                channel.clone(),
                routes,
                DirectDialer,
                all_ports,
                tcp_ports,
                udp_ports,
                netstack_down_tx.clone(),
            ),
        };

        // Pump packets the dataplane routed to this transport up into the forwarder netstack,
        // dropping what its bounded ingress queue has no room for. This forwarder dials IPv4 only,
        // so every IPv6 packet is answered here with a TCP RST / ICMPv6 prohibited (or dropped if
        // it can't be answered safely) and never reaches the netstack, whose wildcard listeners
        // would otherwise accept it. IPv4 goes through the port gate in all-port mode.
        joinset.spawn(async move {
            let mut reject_budget = RejectBudget::new();
            let batching = ts_netstack_smoltcp::netcore::experiment::enabled(
                ts_netstack_smoltcp::netcore::experiment::RX_BATCH,
            );
            let pass = |b: &[u8]| {
                if batching {
                    netstack_down_tx.try_send_deferred(b)
                } else {
                    netstack_down_tx.try_send(b)
                }
            };
            let admit = |b: &[u8]| {
                if b.first().is_some_and(|v| v >> 4 == 6) {
                    // Replies skip the bounded netstack ingress, so cap their rate; over budget the
                    // packet is still consumed, just not answered.
                    if reject_budget.take()
                        && let Some(reply) = ts_forwarder::reject_ipv6(b)
                        && reject_up.send(vec![reply.into()]).is_err()
                    {
                        tracing::debug!("dataplane gone; IPv6 rejection not sent");
                    }
                    return true;
                }
                match &gate {
                    Some(gate) => {
                        (if batching {
                            gate.ingress_with(b, pass)
                        } else {
                            gate.ingress(b)
                        }) != Admit::Dropped
                    }
                    None => pass(b),
                }
            };
            crate::netstack_actor::pump_ingress_with_batch_wake(
                netstack_down,
                admit,
                || {
                    if batching {
                        netstack_down_tx.notify_receive();
                    }
                },
                "forwarder",
                batching,
            )
            .await;

            tracing::warn!("forwarder netstack uplink shut down!");
        });

        tracing::debug!(
            n_routes,
            n_tcp_ports,
            n_udp_ports,
            all_ports,
            exit_egress = env.forward_exit_egress,
            exit_proxy = env.exit_proxy.is_some(),
            exit_node_dns = ?env.exit_node_dns,
            dialer = ?choice,
            "forwarder started"
        );

        Ok(Self {
            _joinset: joinset,
            channel,
            route_updater,
        })
    }
}

#[kameo::messages]
impl ForwarderActor {
    #[message]
    pub fn get_channel(&self) -> (Channel,) {
        (self.channel.clone(),)
    }

    /// Replace the forwarder's accept/dial route table with `routes` (a runtime
    /// `set_advertise_routes`). New flows are classified against the new set; in-flight flows keep
    /// their existing classification. The local half of advertising routes — paired with the wire
    /// half (re-advertising `Hostinfo.RoutableIPs` to control) so the node both forwards and is
    /// granted exactly the prefixes it advertises.
    #[message]
    pub fn update_routes(&self, routes: Vec<ipnet::IpNet>) {
        self.route_updater.update(RouteTable::new(routes));
    }
}

#[cfg(test)]
mod tests {
    /// The IPv6 rejection budget allows a burst of `PER_SEC`, then refuses until it refills.
    #[test]
    fn reject_budget_caps_burst() {
        let mut b = super::RejectBudget::new();
        let allowed = (0..3000).filter(|_| b.take()).count();
        assert!((1000..1010).contains(&allowed), "allowed {allowed}");
    }

    use super::*;
    use crate::env::ForwarderConfig;

    /// Build a `ForwarderConfig` toggling only the two gate bools (the historically swap-prone
    /// adjacent params), leaving everything else fail-closed/empty.
    fn cfg(forward_all_ports: bool, forward_exit_egress: bool) -> ForwarderConfig {
        ForwarderConfig {
            accept_routes: false,
            accept_dns: true,
            exit_node: None,
            forward_routes: vec![],
            forward_tcp_ports: vec![],
            forward_udp_ports: vec![],
            forward_all_ports,
            forward_exit_egress,
            block_incoming: false,
            exit_proxy: None,
            exit_node_dns: None,
            peerapi_port: None,
            taildrop_dir: None,
            enable_ipv6: false,
            wireguard_listen_port: None,
            network_monitor: false,
            persistent_keepalive_interval: None,
            ingress_active: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
        }
    }

    fn proxy_cfg() -> ts_forwarder::ProxyConfig {
        ts_forwarder::ProxyConfig {
            addr: "203.0.113.7:1080".parse().unwrap(),
            scheme: ts_forwarder::ProxyScheme::Socks5,
            auth: None,
        }
    }

    #[test]
    fn host_exit_dialer_iff_forward_exit_egress() {
        // Fail-closed default: no exit egress => the direct (refusing) dialer.
        assert_eq!(
            dialer_choice(cfg(false, false).forward_exit_egress, false),
            DialerChoice::Direct
        );
        // Opt-in: exit egress, no proxy => the host-exit dialer that egresses via the real IP.
        assert_eq!(
            dialer_choice(cfg(false, true).forward_exit_egress, false),
            DialerChoice::HostExit
        );
        // The all-ports flag is orthogonal: it must not affect the dialer gate.
        assert_eq!(
            dialer_choice(cfg(true, false).forward_exit_egress, false),
            DialerChoice::Direct
        );
        assert_eq!(
            dialer_choice(cfg(true, true).forward_exit_egress, false),
            DialerChoice::HostExit
        );
    }

    #[test]
    fn proxy_dialer_iff_exit_egress_and_proxy_configured() {
        // Exit egress on + proxy configured => proxy dialer (egress via the proxy IP).
        assert_eq!(
            dialer_choice(cfg(false, true).forward_exit_egress, true),
            DialerChoice::Proxy
        );
        // A configured proxy with exit egress OFF must NOT enable proxy egress — fail-closed wins,
        // so the real IP can never leak just because a proxy happens to be configured.
        assert_eq!(
            dialer_choice(cfg(false, false).forward_exit_egress, true),
            DialerChoice::Direct
        );
        // Exit egress on, no proxy => host-exit (real IP), proxy dialer only when one is set.
        assert_eq!(
            dialer_choice(cfg(false, true).forward_exit_egress, false),
            DialerChoice::HostExit
        );
    }

    #[test]
    fn exit_proxy_converts_through_control_config() {
        // The transport-only ts_control type round-trips into the ts_forwarder dialer config via
        // ForwarderConfig::from_control_config (the only place ts_control<->ts_forwarder cross).
        let control = ts_control::Config {
            forward_exit_egress: true,
            exit_proxy: Some(ts_control::ExitProxyConfig {
                addr: "198.51.100.9:8080".parse().unwrap(),
                scheme: ts_control::ExitProxyScheme::HttpConnect,
                auth: Some(("user".to_owned(), "pass".to_owned())),
            }),
            ..Default::default()
        };

        let fwd = ForwarderConfig::from_control_config(&control);
        // It selects the proxy dialer.
        assert_eq!(
            dialer_choice(fwd.forward_exit_egress, fwd.exit_proxy.is_some()),
            DialerChoice::Proxy
        );
        let proxy = fwd.exit_proxy.expect("proxy threaded through");
        assert_eq!(proxy.addr, "198.51.100.9:8080".parse().unwrap());
        assert_eq!(proxy.scheme, ts_forwarder::ProxyScheme::HttpConnect);
        assert_eq!(proxy.auth, Some(("user".to_owned(), "pass".to_owned())));
    }

    #[test]
    fn exit_proxy_absent_when_unconfigured() {
        let control = ts_control::Config::default();
        let fwd = ForwarderConfig::from_control_config(&control);
        assert!(fwd.exit_proxy.is_none());
        // Touch the helper constructor so it's covered and the unused-fn lint stays quiet.
        let cfg = proxy_cfg();
        assert_eq!(cfg.scheme, ts_forwarder::ProxyScheme::Socks5);
    }

    #[test]
    fn all_ports_mode_iff_forward_all_ports() {
        assert_eq!(
            forwarder_mode(cfg(false, false).forward_all_ports),
            ForwarderMode::Ports
        );
        assert_eq!(
            forwarder_mode(cfg(true, false).forward_all_ports),
            ForwarderMode::AllPorts
        );
        // Orthogonal to the exit-egress gate.
        assert_eq!(
            forwarder_mode(cfg(false, true).forward_all_ports),
            ForwarderMode::Ports
        );
        assert_eq!(
            forwarder_mode(cfg(true, true).forward_all_ports),
            ForwarderMode::AllPorts
        );
    }
}