feat(vmm): add userspace port forwarding for bridge-mode VMs

Introduce dstack-port-forward crate with TCP (splice zero-copy with
fallback) and UDP forwarding. Integrate ForwardService into VMM to
dynamically manage port forwarding rules when bridge-mode guests
report their eth0 IP via the network.eth0 vsock event.

Key changes:
- New port-forward crate with ForwardService for dynamic rule add/remove
- Guest reports eth0 IP in dstack-prepare.sh via dstack-util notify-host
- VMM persists guest IP to disk and hot-reloads forwarding rules
- On VMM restart, forwarding is restored from persisted guest IPs
- guest-agent now includes eth0 interfaces in network info API

Author: kvinwang

Cargo.lock

@@ -52,6 +52,7 @@ members = [
     "verifier",
     "no_std_check",
     "size-parser",
+    "port-forward",
 ]
 resolver = "2"
 
@@ -64,6 +65,7 @@ dstack-gateway-rpc = { path = "gateway/rpc" }
 dstack-kms-rpc = { path = "kms/rpc" }
 dstack-guest-agent-rpc = { path = "guest-agent/rpc" }
 dstack-vmm-rpc = { path = "vmm/rpc" }
+dstack-port-forward = { path = "port-forward" }
 cc-eventlog = { path = "cc-eventlog" }
 supervisor = { path = "supervisor" }
 supervisor-client = { path = "supervisor/client" }

Cargo.toml

@@ -278,6 +278,13 @@ echo "============================"
 
 cd /dstack
 
+# Report eth0 IP address to VMM for bridge-mode port forwarding
+ETH0_IP=$(ip -4 -o addr show eth0 2>/dev/null | awk '{print $4}' | cut -d/ -f1 || true)
+if [ -n "$ETH0_IP" ]; then
+	log "Reporting eth0 IP: $ETH0_IP"
+	dstack-util notify-host -e "network.eth0" -d "$ETH0_IP" || true
+fi
+
 if [ $(jq 'has("init_script")' app-compose.json) == true ]; then
 	log "Running init script"
 	dstack-util notify-host -e "boot.progress" -d "init-script" || true

basefiles/dstack-prepare.sh

@@ -145,8 +145,11 @@ fn get_interfaces() -> Vec<Interface> {
     sysinfo::Networks::new_with_refreshed_list()
         .into_iter()
         .filter_map(|(interface_name, network)| {
-            if !(interface_name == "dstack-wg0" || interface_name.starts_with("enp")) {
-                // We only get dstack-wg0 and enp interfaces.
+            if !(interface_name == "dstack-wg0"
+                || interface_name.starts_with("enp")
+                || interface_name.starts_with("eth"))
+            {
+                // We only get dstack-wg0, enp and eth interfaces.
                 // Docker bridge is not included due to privacy concerns.
                 return None;
             }

guest-agent/src/guest_api_service.rs

@@ -0,0 +1,14 @@
+[package]
+name = "dstack-port-forward"
+version.workspace = true
+edition.workspace = true
+
+[dependencies]
+anyhow.workspace = true
+tracing.workspace = true
+tokio = { workspace = true, features = ["net", "rt", "io-util", "macros", "sync", "time"] }
+tokio-util = { version = "0.7", features = ["rt"] }
+nix = { workspace = true, features = ["fs", "net", "zerocopy"] }
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["full"] }

port-forward/Cargo.toml

@@ -0,0 +1,124 @@
+use std::collections::HashMap;
+use std::net::{IpAddr, SocketAddr};
+
+use anyhow::{bail, Result};
+use tokio::task::JoinHandle;
+use tokio_util::sync::CancellationToken;
+
+mod tcp;
+mod udp;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum Protocol {
+    Tcp,
+    Udp,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct ForwardRule {
+    pub protocol: Protocol,
+    pub listen_addr: IpAddr,
+    pub listen_port: u16,
+    pub target_ip: IpAddr,
+    pub target_port: u16,
+}
+
+impl ForwardRule {
+    fn listen_sock(&self) -> SocketAddr {
+        SocketAddr::new(self.listen_addr, self.listen_port)
+    }
+
+    fn target_sock(&self) -> SocketAddr {
+        SocketAddr::new(self.target_ip, self.target_port)
+    }
+}
+
+struct RunningRule {
+    cancel: CancellationToken,
+    task: JoinHandle<()>,
+}
+
+/// Manages a dynamic set of port forwarding rules.
+///
+/// Rules can be added and removed at runtime. Dropping the service cancels all
+/// forwarding tasks.
+pub struct ForwardService {
+    cancel: CancellationToken,
+    rules: HashMap<ForwardRule, RunningRule>,
+}
+
+impl ForwardService {
+    pub fn new() -> Self {
+        Self {
+            cancel: CancellationToken::new(),
+            rules: HashMap::new(),
+        }
+    }
+
+    /// Add a forwarding rule. Returns error if the rule already exists.
+    pub fn add_rule(&mut self, rule: ForwardRule) -> Result<()> {
+        if self.rules.contains_key(&rule) {
+            bail!("rule already exists: {:?}", rule);
+        }
+
+        let token = self.cancel.child_token();
+        let listen = rule.listen_sock();
+        let target = rule.target_sock();
+
+        let task = match rule.protocol {
+            Protocol::Tcp => tokio::spawn(tcp::run_tcp_forwarder(listen, target, token.clone())),
+            Protocol::Udp => tokio::spawn(udp::run_udp_forwarder(listen, target, token.clone())),
+        };
+
+        tracing::info!("added forwarding rule: {listen} -> {target} ({:?})", rule.protocol);
+        self.rules.insert(rule, RunningRule { cancel: token, task });
+        Ok(())
+    }
+
+    /// Remove a forwarding rule and stop its task.
+    pub async fn remove_rule(&mut self, rule: &ForwardRule) -> Result<()> {
+        match self.rules.remove(rule) {
+            Some(running) => {
+                running.cancel.cancel();
+                let _ = running.task.await;
+                tracing::info!(
+                    "removed forwarding rule: {} -> {} ({:?})",
+                    rule.listen_sock(),
+                    rule.target_sock(),
+                    rule.protocol,
+                );
+                Ok(())
+            }
+            None => bail!("rule not found: {:?}", rule),
+        }
+    }
+
+    /// Number of active rules.
+    pub fn len(&self) -> usize {
+        self.rules.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.rules.is_empty()
+    }
+
+    /// Gracefully stop all forwarding and wait for tasks to finish.
+    pub async fn shutdown(mut self) {
+        self.cancel.cancel();
+        for (_, running) in std::mem::take(&mut self.rules) {
+            let _ = running.task.await;
+        }
+    }
+}
+
+impl Default for ForwardService {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Drop for ForwardService {
+    fn drop(&mut self) {
+        self.cancel.cancel();
+    }
+}