diff --git a/docs/bridge-networking.md b/docs/bridge-networking.md index a96d5c9dc..0d16fd476 100644 --- a/docs/bridge-networking.md +++ b/docs/bridge-networking.md @@ -90,13 +90,6 @@ sudo sysctl -p /etc/sysctl.d/99-dstack-bridge.conf sudo apt install -y dnsmasq ``` -Install the DHCP notification script (notifies VMM when a VM gets an IP so port forwarding can be established): - -```bash -sudo cp dstack/scripts/dhcp-notify.sh /usr/local/bin/dhcp-notify.sh -sudo chmod +x /usr/local/bin/dhcp-notify.sh -``` - Create dnsmasq config: ```ini @@ -106,11 +99,8 @@ bind-interfaces dhcp-range=10.0.100.10,10.0.100.254,255.255.255.0,12h dhcp-option=option:router,10.0.100.1 dhcp-option=option:dns-server,8.8.8.8,1.1.1.1 -dhcp-script=/usr/local/bin/dhcp-notify.sh ``` -The `dhcp-script` option tells dnsmasq to call the notification script on every lease event. The script sends the MAC and IP to VMM's `ReportDhcpLease` RPC, which triggers automatic port forwarding for the VM. - ```bash sudo systemctl restart dnsmasq ``` @@ -172,8 +162,7 @@ sudo chmod u+s /usr/lib/qemu/qemu-bridge-helper - VMM passes `-netdev bridge,id=net0,br=` to QEMU - QEMU's bridge helper (setuid) creates a TAP device and attaches it to the bridge - Guest MAC address is derived from SHA256 of the VM ID, with an optional configurable prefix (stable across restarts for DHCP IP consistency) -- The host DHCP server (dnsmasq) assigns an IP and calls `dhcp-notify.sh`, which notifies VMM via the `ReportDhcpLease` RPC -- VMM matches the MAC address to identify the VM and establishes port forwarding rules +- The host DHCP server (dnsmasq) assigns an IP to the VM - When QEMU exits, the TAP device is automatically destroyed - VMM does not need root or `CAP_NET_ADMIN` diff --git a/dstack/Cargo.lock b/dstack/Cargo.lock index 6b439ce00..62d74866a 100644 --- a/dstack/Cargo.lock +++ b/dstack/Cargo.lock @@ -1900,17 +1900,6 @@ dependencies = [ "tracing-subscriber", ] -[[package]] -name = "dstack-port-forward" -version = "0.6.0" -dependencies = [ - "anyhow", - "nix 0.29.0", - "tokio", - "tokio-util", - "tracing", -] - [[package]] name = "dstack-tee-simulator" version = "0.6.0" @@ -2043,7 +2032,6 @@ dependencies = [ "dirs 6.0.0", "dstack-kms-rpc", "dstack-mr", - "dstack-port-forward", "dstack-types", "dstack-vmm-rpc", "fatfs", @@ -4066,7 +4054,6 @@ dependencies = [ "cfg-if", "cfg_aliases", "libc", - "memoffset 0.9.1", ] [[package]] @@ -7105,7 +7092,6 @@ dependencies = [ "bytes", "futures-core", "futures-sink", - "futures-util", "pin-project-lite", "tokio", ] diff --git a/dstack/Cargo.toml b/dstack/Cargo.toml index 1fd8daeeb..e1d663951 100644 --- a/dstack/Cargo.toml +++ b/dstack/Cargo.toml @@ -61,7 +61,6 @@ members = [ "dstack-mr/cli", "verifier", "size-parser", - "port-forward", "crates/dstack-cli-core", "crates/dstack-cli", "crates/dstackup", @@ -78,7 +77,6 @@ dstack-kms-rpc = { path = "kms/rpc" } dstack-guest-agent-rpc = { path = "guest-agent/rpc" } dstack-vmm-rpc = { path = "vmm/rpc" } dstack-cli-core = { path = "crates/dstack-cli-core" } -dstack-port-forward = { path = "port-forward" } cc-eventlog = { path = "cc-eventlog" } supervisor = { path = "supervisor" } supervisor-client = { path = "supervisor/client" } diff --git a/dstack/crates/dstack-cli-core/src/config.rs b/dstack/crates/dstack-cli-core/src/config.rs index a0ca82417..dd0219f26 100644 --- a/dstack/crates/dstack-cli-core/src/config.rs +++ b/dstack/crates/dstack-cli-core/src/config.rs @@ -352,7 +352,6 @@ mode = "user" net = "10.0.2.0/24" dhcp_start = "10.0.2.10" restrict = false -forward_service_enabled = false [cvm.port_mapping] enabled = true diff --git a/dstack/port-forward/Cargo.toml b/dstack/port-forward/Cargo.toml deleted file mode 100644 index 5422080d5..000000000 --- a/dstack/port-forward/Cargo.toml +++ /dev/null @@ -1,18 +0,0 @@ -# SPDX-FileCopyrightText: © 2024-2025 Phala Network -# -# SPDX-License-Identifier: Apache-2.0 - -[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"] } diff --git a/dstack/port-forward/src/lib.rs b/dstack/port-forward/src/lib.rs deleted file mode 100644 index beeb87d9a..000000000 --- a/dstack/port-forward/src/lib.rs +++ /dev/null @@ -1,137 +0,0 @@ -// SPDX-FileCopyrightText: © 2024-2025 Phala Network -// -// SPDX-License-Identifier: Apache-2.0 - -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, -} - -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(); - } -} diff --git a/dstack/port-forward/src/tcp.rs b/dstack/port-forward/src/tcp.rs deleted file mode 100644 index 50878552f..000000000 --- a/dstack/port-forward/src/tcp.rs +++ /dev/null @@ -1,207 +0,0 @@ -// SPDX-FileCopyrightText: © 2024-2025 Phala Network -// -// SPDX-License-Identifier: Apache-2.0 - -use std::net::SocketAddr; -use std::os::unix::io::{AsFd, AsRawFd, BorrowedFd, OwnedFd}; - -use tokio::io; -use tokio::net::{TcpListener, TcpStream}; -use tokio_util::sync::CancellationToken; - -/// Run a TCP port forwarder: listen on `listen_addr`, forward to `target`. -pub async fn run_tcp_forwarder( - listen_addr: SocketAddr, - target: SocketAddr, - cancel: CancellationToken, -) { - let listener = match TcpListener::bind(listen_addr).await { - Ok(l) => l, - Err(e) => { - tracing::error!("tcp bind {listen_addr} failed: {e}"); - return; - } - }; - tracing::info!("tcp forwarding {listen_addr} -> {target}"); - - loop { - tokio::select! { - _ = cancel.cancelled() => break, - result = listener.accept() => { - let (client, peer) = match result { - Ok(v) => v, - Err(e) => { - tracing::warn!("tcp accept on {listen_addr}: {e}"); - continue; - } - }; - let cancel = cancel.child_token(); - tokio::spawn(handle_tcp_connection(client, peer, target, cancel)); - } - } - } -} - -async fn handle_tcp_connection( - mut client: TcpStream, - peer: SocketAddr, - target: SocketAddr, - cancel: CancellationToken, -) { - let mut server = match TcpStream::connect(target).await { - Ok(s) => s, - Err(e) => { - tracing::debug!("tcp connect to {target} for {peer}: {e}"); - return; - } - }; - - let _ = client.set_nodelay(true); - let _ = server.set_nodelay(true); - - let result = tokio::select! { - _ = cancel.cancelled() => return, - r = relay(&mut client, &mut server) => r, - }; - - if let Err(e) = result { - tracing::debug!("tcp relay {peer} <-> {target}: {e}"); - } -} - -async fn relay(client: &mut TcpStream, server: &mut TcpStream) -> io::Result<()> { - // Try splice(2) zero-copy first, fall back to userspace copy. - match splice_bidirectional(client, server).await { - Ok(()) => Ok(()), - Err(e) if e.kind() == io::ErrorKind::Unsupported => { - tracing::debug!("splice not supported, falling back to copy_bidirectional"); - io::copy_bidirectional(client, server).await?; - Ok(()) - } - Err(e) => Err(e), - } -} - -/// Zero-copy bidirectional TCP relay using Linux splice(2). -/// -/// When one direction hits EOF, select! drops the other direction. -async fn splice_bidirectional(a: &TcpStream, b: &TcpStream) -> io::Result<()> { - let a_fd = a.as_raw_fd(); - let b_fd = b.as_raw_fd(); - - tokio::select! { - r = splice_one_direction(a, a_fd, b, b_fd) => r, - r = splice_one_direction(b, b_fd, a, a_fd) => r, - } -} - -/// Splice data from src fd to dst fd via an intermediate pipe. -/// -/// Uses `TcpStream::try_io` for proper readiness handling: when splice returns -/// EAGAIN, try_io automatically clears the readiness flag so the next -/// `readable().await` / `writable().await` blocks until the fd is truly ready. -async fn splice_one_direction( - src: &TcpStream, - src_fd: i32, - dst: &TcpStream, - dst_fd: i32, -) -> io::Result<()> { - use nix::fcntl::{splice, SpliceFFlags}; - use nix::unistd::pipe; - - let (pipe_r, pipe_w): (OwnedFd, OwnedFd) = pipe().map_err(io::Error::other)?; - - let flags = SpliceFFlags::SPLICE_F_NONBLOCK | SpliceFFlags::SPLICE_F_MOVE; - let chunk: usize = 65536; - - let src_bfd = unsafe { BorrowedFd::borrow_raw(src_fd) }; - let dst_bfd = unsafe { BorrowedFd::borrow_raw(dst_fd) }; - - loop { - src.readable().await?; - - let n = match src.try_io(io::Interest::READABLE, || { - match splice(src_bfd, None, pipe_w.as_fd(), None, chunk, flags) { - Ok(n) => Ok(n), - Err(nix::errno::Errno::EAGAIN) => Err(io::ErrorKind::WouldBlock.into()), - Err(nix::errno::Errno::EINVAL) => Err(io::Error::new( - io::ErrorKind::Unsupported, - "splice not supported for this fd type", - )), - Err(e) => Err(io::Error::other(e)), - } - }) { - Ok(0) => return Ok(()), - Ok(n) => n, - Err(e) if e.kind() == io::ErrorKind::WouldBlock => continue, - Err(e) => return Err(e), - }; - - let mut written = 0; - while written < n { - dst.writable().await?; - match dst.try_io(io::Interest::WRITABLE, || { - match splice(pipe_r.as_fd(), None, dst_bfd, None, n - written, flags) { - Ok(w) => Ok(w), - Err(nix::errno::Errno::EAGAIN) => Err(io::ErrorKind::WouldBlock.into()), - Err(e) => Err(io::Error::other(e)), - } - }) { - Ok(w) => written += w, - Err(e) if e.kind() == io::ErrorKind::WouldBlock => continue, - Err(e) => return Err(e), - } - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - - #[tokio::test] - async fn test_tcp_forward_roundtrip() { - // Echo server - let echo = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let echo_addr = echo.local_addr().unwrap(); - tokio::spawn(async move { - loop { - let (mut conn, _) = echo.accept().await.unwrap(); - tokio::spawn(async move { - let mut buf = vec![0u8; 4096]; - loop { - let n = match conn.read(&mut buf).await { - Ok(0) | Err(_) => break, - Ok(n) => n, - }; - if conn.write_all(&buf[..n]).await.is_err() { - break; - } - } - }); - } - }); - - // Start forwarder on a free port - let tmp = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let fwd_addr = tmp.local_addr().unwrap(); - drop(tmp); - - let cancel = CancellationToken::new(); - tokio::spawn(run_tcp_forwarder(fwd_addr, echo_addr, cancel.child_token())); - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - - let mut conn = TcpStream::connect(fwd_addr).await.unwrap(); - conn.write_all(b"hello splice").await.unwrap(); - - let mut buf = vec![0u8; 64]; - let n = tokio::time::timeout(std::time::Duration::from_secs(2), conn.read(&mut buf)) - .await - .unwrap() - .unwrap(); - assert_eq!(&buf[..n], b"hello splice"); - - cancel.cancel(); - } -} diff --git a/dstack/port-forward/src/udp.rs b/dstack/port-forward/src/udp.rs deleted file mode 100644 index 78b7af5bd..000000000 --- a/dstack/port-forward/src/udp.rs +++ /dev/null @@ -1,201 +0,0 @@ -// SPDX-FileCopyrightText: © 2024-2025 Phala Network -// -// SPDX-License-Identifier: Apache-2.0 - -use std::collections::HashMap; -use std::net::SocketAddr; -use std::sync::Arc; -use std::time::{Duration, Instant}; - -use tokio::net::UdpSocket; -use tokio::sync::Mutex; -use tokio_util::sync::CancellationToken; - -const UDP_IDLE_TIMEOUT: Duration = Duration::from_secs(60); -const UDP_BUF_SIZE: usize = 65535; -const CLEANUP_INTERVAL: Duration = Duration::from_secs(10); - -struct ClientState { - /// Ephemeral socket for communicating with the target on behalf of this client. - socket: Arc, - last_active: Instant, - /// Cancel token for the return-path task. - _cancel: CancellationToken, -} - -/// Run a UDP port forwarder: listen on `listen_addr`, forward to `target`. -pub async fn run_udp_forwarder( - listen_addr: SocketAddr, - target: SocketAddr, - cancel: CancellationToken, -) { - let host_socket = match UdpSocket::bind(listen_addr).await { - Ok(s) => s, - Err(e) => { - tracing::error!("udp bind {listen_addr} failed: {e}"); - return; - } - }; - tracing::info!("udp forwarding {listen_addr} -> {target}"); - - let host_socket = Arc::new(host_socket); - let clients: Arc>> = - Arc::new(Mutex::new(HashMap::new())); - - // Periodic cleanup of idle client entries - let cleanup_clients = clients.clone(); - let cleanup_cancel = cancel.child_token(); - tokio::spawn(async move { - loop { - tokio::select! { - _ = cleanup_cancel.cancelled() => break, - _ = tokio::time::sleep(CLEANUP_INTERVAL) => { - let mut map = cleanup_clients.lock().await; - let now = Instant::now(); - map.retain(|addr, entry| { - let alive = now.duration_since(entry.last_active) < UDP_IDLE_TIMEOUT; - if !alive { - tracing::debug!("udp client {addr} idle timeout"); - } - alive - // Dropping ClientState cancels the return-path task via _cancel - }); - } - } - } - }); - - let mut buf = vec![0u8; UDP_BUF_SIZE]; - - loop { - tokio::select! { - _ = cancel.cancelled() => break, - result = host_socket.recv_from(&mut buf) => { - let (n, client_addr) = match result { - Ok(v) => v, - Err(e) => { - tracing::warn!("udp recv on {listen_addr}: {e}"); - continue; - } - }; - - let data = &buf[..n]; - let mut map = clients.lock().await; - - // Get or create per-client ephemeral socket - let entry = match map.entry(client_addr) { - std::collections::hash_map::Entry::Occupied(e) => e.into_mut(), - std::collections::hash_map::Entry::Vacant(e) => { - let socket = match (|| -> anyhow::Result> { - let std_sock = std::net::UdpSocket::bind("0.0.0.0:0")?; - std_sock.set_nonblocking(true)?; - Ok(Arc::new(UdpSocket::from_std(std_sock)?)) - })() { - Ok(s) => s, - Err(e) => { - tracing::warn!("udp ephemeral socket for {client_addr}: {e}"); - continue; - } - }; - - let return_cancel = cancel.child_token(); - tokio::spawn(udp_return_path( - host_socket.clone(), - socket.clone(), - client_addr, - return_cancel.child_token(), - )); - - e.insert(ClientState { - socket, - last_active: Instant::now(), - _cancel: return_cancel, - }) - } - }; - entry.last_active = Instant::now(); - - // Forward client data to target - if let Err(e) = entry.socket.send_to(data, target).await { - tracing::debug!("udp send to {target} for {client_addr}: {e}"); - } - } - } - } -} - -/// Return path: read from the per-client ephemeral socket, send back to the -/// original client via the host socket. -async fn udp_return_path( - host_socket: Arc, - client_socket: Arc, - client_addr: SocketAddr, - cancel: CancellationToken, -) { - let mut buf = vec![0u8; UDP_BUF_SIZE]; - - loop { - tokio::select! { - _ = cancel.cancelled() => break, - result = client_socket.recv_from(&mut buf) => { - let (n, _from) = match result { - Ok(v) => v, - Err(e) => { - tracing::debug!("udp return recv for {client_addr}: {e}"); - break; - } - }; - if let Err(e) = host_socket.send_to(&buf[..n], client_addr).await { - tracing::debug!("udp return send to {client_addr}: {e}"); - } - } - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn test_udp_forward_roundtrip() { - // UDP echo server - let echo = UdpSocket::bind("127.0.0.1:0").await.unwrap(); - let echo_addr = echo.local_addr().unwrap(); - tokio::spawn(async move { - let mut buf = vec![0u8; UDP_BUF_SIZE]; - loop { - match echo.recv_from(&mut buf).await { - Ok((n, from)) => { - let _ = echo.send_to(&buf[..n], from).await; - } - Err(_) => break, - } - } - }); - - // Start forwarder - let cancel = CancellationToken::new(); - // Bind to get a free port, then release and let forwarder bind - let tmp = UdpSocket::bind("127.0.0.1:0").await.unwrap(); - let fwd_addr = tmp.local_addr().unwrap(); - drop(tmp); - - let token = cancel.child_token(); - tokio::spawn(run_udp_forwarder(fwd_addr, echo_addr, token)); - tokio::time::sleep(Duration::from_millis(50)).await; - - // Send through forwarder - let client = UdpSocket::bind("127.0.0.1:0").await.unwrap(); - client.send_to(b"hello udp", fwd_addr).await.unwrap(); - - let mut buf = vec![0u8; 64]; - let (n, _) = tokio::time::timeout(Duration::from_secs(2), client.recv_from(&mut buf)) - .await - .unwrap() - .unwrap(); - assert_eq!(&buf[..n], b"hello udp"); - - cancel.cancel(); - } -} diff --git a/dstack/scripts/dhcp-notify.sh b/dstack/scripts/dhcp-notify.sh deleted file mode 100755 index 697b3dd13..000000000 --- a/dstack/scripts/dhcp-notify.sh +++ /dev/null @@ -1,38 +0,0 @@ -#!/bin/bash -# SPDX-FileCopyrightText: © 2025 Phala Network -# SPDX-License-Identifier: Apache-2.0 -# -# DHCP lease notification script for dnsmasq. -# -# Called by dnsmasq via --dhcp-script on lease events. -# Notifies dstack-vmm of the guest's MAC and IP so that port -# forwarding can be established automatically. -# -# Arguments (set by dnsmasq): -# $1 action add | del | old -# $2 mac MAC address of the guest NIC -# $3 ip IPv4 address assigned by DHCP -# $4 hostname (optional) -# -# Configuration: -# VMM_URL Base URL of dstack-vmm (default: http://127.0.0.1:9080) - -ACTION="$1" -MAC="$2" -IP="$3" - -VMM_URL="${VMM_URL:-http://127.0.0.1:9080}" - -logger -t dhcp-notify "action=$ACTION mac=$MAC ip=$IP" - -case "$ACTION" in - add|old) - curl -s -X POST "${VMM_URL}/prpc/ReportDhcpLease" \ - -H 'Content-Type: application/json' \ - -d "{\"mac\":\"$MAC\",\"ip\":\"$IP\"}" \ - || logger -t dhcp-notify "failed to notify VMM" - ;; - del) - # Could clear forwarding on lease expiry; not implemented yet. - ;; -esac diff --git a/dstack/scripts/setup-bridge.sh b/dstack/scripts/setup-bridge.sh index f6fbf2247..edcbe35a7 100755 --- a/dstack/scripts/setup-bridge.sh +++ b/dstack/scripts/setup-bridge.sh @@ -192,47 +192,6 @@ check_dhcp() { check_info "Run: $(basename "$0") setup --mode standalone --bridge $BRIDGE" } -check_dhcp_notify() { - echo - bold "DHCP lease notification" - - local provider - provider=$(detect_bridge_provider) - - if [[ "$provider" == libvirt:* ]]; then - check_warn "libvirt DHCP does not support dhcp-script callback" - check_info "port forwarding requires manual PRPC call or alternative notification" - return - fi - - # Check dnsmasq config for dhcp-script - local conf_files=(/etc/dnsmasq.d/*"$BRIDGE"* /etc/dnsmasq.d/*.conf) - local found_script="" - for f in "${conf_files[@]}"; do - [[ -f "$f" ]] || continue - local script_path - script_path=$(grep -oP '^dhcp-script=\K.*' "$f" 2>/dev/null || true) - if [[ -n "$script_path" ]]; then - found_script="$script_path" - break - fi - done - - if [[ -z "$found_script" ]]; then - check_warn "no dhcp-script configured in dnsmasq" - check_info "port forwarding will not be set up automatically" - check_info "add 'dhcp-script=/usr/local/bin/dhcp-notify.sh' to dnsmasq config" - return - fi - - if [[ -x "$found_script" ]]; then - check_pass "dhcp-script configured: $found_script" - else - check_fail "dhcp-script $found_script is not executable or missing" - check_info "Fix: sudo chmod +x $found_script" - fi -} - check_ip_forward() { echo bold "IP forwarding" @@ -542,28 +501,12 @@ sudo mv /tmp/.dstack-br-network $network" dhcp_end="${prefix}.254" fi - # Install dhcp-notify.sh if present - local notify_script="/usr/local/bin/dhcp-notify.sh" - local dhcp_script_line="" - local src_notify - src_notify="$(cd "$(dirname "$0")" && pwd)/dhcp-notify.sh" - if [[ -f "$src_notify" ]]; then - run_cmd sudo cp "$src_notify" "$notify_script" - run_cmd sudo chmod +x "$notify_script" - dhcp_script_line="dhcp-script=${notify_script}" - echo " installed $notify_script" - else - echo " $(yellow '[WARN]') dhcp-notify.sh not found at $src_notify" - echo " VM port forwarding will not be set up automatically" - fi - run_cmd bash -c "cat > /tmp/.dstack-dnsmasq <, pub supervisor: SupervisorClient, state: Arc>, - forward_service: Arc>, /// Pull status for registry images: tag → status. pub(crate) pull_status: Arc>>, } @@ -273,10 +271,8 @@ impl App { state: Arc::new(Mutex::new(AppState { cid_pool, vms: HashMap::new(), - active_forwards: HashMap::new(), })), config: Arc::new(config), - forward_service: Arc::new(tokio::sync::Mutex::new(ForwardService::new())), pull_status: Arc::new(Mutex::new(std::collections::HashMap::new())), } } @@ -425,7 +421,6 @@ impl App { pub async fn stop_vm(&self, id: &str) -> Result<()> { self.set_started(id, false)?; - self.cleanup_port_forward(id).await; self.supervisor.stop(id).await?; Ok(()) } @@ -447,9 +442,6 @@ impl App { warn!("failed to write .removing marker for {id}: {err:?}"); } - // Clean up port forwarding immediately - self.cleanup_port_forward(id).await; - // User-initiated removal always deletes the workdir let app = self.clone(); let id = id.to_string(); @@ -560,140 +552,6 @@ impl App { true } - /// Handle a DHCP lease notification: look up VM by MAC address, persist - /// the guest IP, and reconfigure port forwarding. - pub async fn report_dhcp_lease(&self, mac: &str, ip: &str) { - use crate::app::network::mac_address_for_vm_index; - - let vm_id = { - let mut state = self.lock(); - let found = state.vms.iter_mut().find_map(|(id, vm)| { - let networks = vm.effective_networks(&self.config.cvm); - networks.iter().enumerate().find_map(|(index, networking)| { - let prefix = networking.mac_prefix_bytes(); - let nic_mac = mac_address_for_vm_index(&vm.config.manifest.id, &prefix, index); - (nic_mac == mac).then(|| id.clone()) - }) - }); - let Some(vm_id) = found else { - debug!(mac, ip, "DHCP lease for unknown MAC, ignoring"); - return; - }; - let Some(vm) = state.get_mut(&vm_id) else { - debug!(mac, ip, id = %vm_id, "DHCP lease for missing VM, ignoring"); - return; - }; - let workdir = VmWorkDir::new(vm.config.workdir.clone()); - if let Err(e) = workdir.set_guest_ip(ip) { - error!(mac, ip, "failed to persist guest IP: {e}"); - } - vm.state.guest_ip = ip.to_string(); - info!(mac, ip, id = %vm_id, "DHCP lease updated"); - vm_id - }; - self.reconfigure_port_forward(&vm_id).await; - } - - /// Reconfigure port forwarding for a bridge-mode VM. - /// - /// Computes desired rules from the VM's port_map and guest_ip, then diffs - /// against currently active rules. Only changed rules are added/removed so - /// existing connections on unchanged rules are not interrupted. - pub async fn reconfigure_port_forward(&self, id: &str) { - let info = { - let state = self.lock(); - let Some(vm) = state.get(id) else { - return; - }; - let networks = vm.effective_networks(&self.config.cvm); - if networks - .iter() - .any(|networking| networking.mode == crate::config::NetworkingMode::User) - { - return; - } - if !networks - .iter() - .any(|networking| networking.is_bridge() && networking.forward_service_enabled) - { - return; - } - let guest_ip = vm.state.guest_ip.clone(); - let port_map = vm.config.manifest.port_map.clone(); - (guest_ip, port_map) - }; - - let (guest_ip_str, port_map) = info; - if guest_ip_str.is_empty() { - return; - } - let Ok(guest_ip) = guest_ip_str.parse::() else { - warn!(id, ip = %guest_ip_str, "invalid guest IP, skipping port forward"); - return; - }; - - let new_rules: Vec = port_map - .iter() - .map(|pm| ForwardRule { - protocol: match pm.protocol { - Protocol::Tcp => FwdProtocol::Tcp, - Protocol::Udp => FwdProtocol::Udp, - }, - listen_addr: pm.address, - listen_port: pm.from, - target_ip: guest_ip, - target_port: pm.to, - }) - .collect(); - - let old_rules = self - .lock() - .active_forwards - .get(id) - .cloned() - .unwrap_or_default(); - - let old_set: HashSet<_> = old_rules.iter().collect(); - let new_set: HashSet<_> = new_rules.iter().collect(); - - let mut fwd = self.forward_service.lock().await; - - // Remove rules no longer needed - for rule in old_rules.iter().filter(|r| !new_set.contains(r)) { - if let Err(e) = fwd.remove_rule(rule).await { - warn!(id, ?rule, "failed to remove forwarding rule: {e}"); - } - } - - // Add new rules - for rule in new_rules.iter().filter(|r| !old_set.contains(r)) { - if let Err(e) = fwd.add_rule(rule.clone()) { - warn!(id, ?rule, "failed to add forwarding rule: {e}"); - } - } - - drop(fwd); - self.lock() - .active_forwards - .insert(id.to_string(), new_rules); - info!(id, "port forwarding reconfigured"); - } - - /// Remove all port forwarding rules for a VM. - pub async fn cleanup_port_forward(&self, id: &str) { - let old_rules = self.lock().active_forwards.remove(id).unwrap_or_default(); - if old_rules.is_empty() { - return; - } - let mut fwd = self.forward_service.lock().await; - for rule in &old_rules { - if let Err(e) = fwd.remove_rule(rule).await { - warn!(id, ?rule, "failed to remove forwarding rule: {e}"); - } - } - info!(id, count = old_rules.len(), "port forwarding cleaned up"); - } - pub async fn reload_vms(&self) -> Result<()> { let vm_path = self.vm_dir(); let running_vms = self.supervisor.list().await.context("Failed to list VMs")?; @@ -754,21 +612,6 @@ impl App { } } - // Restore port forwarding for running bridge-mode VMs with persisted guest IPs - let vm_ids: Vec = self.lock().vms.keys().cloned().collect(); - for id in vm_ids { - let workdir = self.work_dir(&id); - if let Some(ip) = workdir.guest_ip() { - { - let mut state = self.lock(); - if let Some(vm) = state.get_mut(&id) { - vm.state.guest_ip = ip; - } - } - self.reconfigure_port_forward(&id).await; - } - } - Ok(()) } @@ -1553,7 +1396,6 @@ mod tests { manifest.networks = vec![Networking { mode: NetworkingMode::Bridge, bridge: "dstack-br0".to_string(), - forward_service_enabled: true, mac_prefix: String::new(), net: String::new(), dhcp_start: String::new(), @@ -1768,7 +1610,6 @@ mod tests { bridge_manifest.networks = vec![Networking { mode: NetworkingMode::Bridge, bridge: "dstack-br0".to_string(), - forward_service_enabled: true, mac_prefix: "02:aa:bb".to_string(), net: String::new(), dhcp_start: String::new(), @@ -2069,7 +1910,6 @@ struct VmStateMut { boot_progress: String, boot_error: String, shutdown_progress: String, - guest_ip: String, runtime_networks: Vec, devices: GpuConfig, events: VecDeque, @@ -2102,21 +1942,11 @@ impl VmState { state: VmStateMut::default(), } } - - pub fn effective_networks(&self, cvm: &crate::config::CvmConfig) -> Vec { - if self.state.runtime_networks.is_empty() { - resolved_networks(&self.config.manifest, cvm) - } else { - self.state.runtime_networks.clone() - } - } } pub(crate) struct AppState { cid_pool: IdPool, vms: HashMap, - /// Tracks active port forwarding rules per VM ID (bridge mode only). - active_forwards: HashMap>, } impl AppState { diff --git a/dstack/vmm/src/app/network.rs b/dstack/vmm/src/app/network.rs index 2ed379423..87925d9a6 100644 --- a/dstack/vmm/src/app/network.rs +++ b/dstack/vmm/src/app/network.rs @@ -63,13 +63,6 @@ pub(crate) fn validate_resolved_network(networking: &Networking) -> Result<()> { } pub(crate) fn validate_resolved_networks(networks: &[Networking]) -> Result<()> { - let forwarding_bridges = networks - .iter() - .filter(|networking| networking.is_bridge() && networking.forward_service_enabled) - .count(); - if forwarding_bridges > 1 { - bail!("built-in port forwarding supports only one bridge"); - } for networking in networks { validate_resolved_network(networking)?; } diff --git a/dstack/vmm/src/app/workdir.rs b/dstack/vmm/src/app/workdir.rs index 6984cd4b5..d5fdc338f 100644 --- a/dstack/vmm/src/app/workdir.rs +++ b/dstack/vmm/src/app/workdir.rs @@ -127,25 +127,10 @@ impl VmWorkDir { self.shared_dir().join(INSTANCE_INFO) } - pub fn guest_ip_path(&self) -> PathBuf { - self.workdir.join("guest-ip") - } - pub fn runtime_networks_path(&self) -> PathBuf { self.workdir.join("runtime-networks.json") } - pub fn guest_ip(&self) -> Option { - fs::read_to_string(self.guest_ip_path()) - .ok() - .map(|s| s.trim().to_string()) - .filter(|s| !s.is_empty()) - } - - pub fn set_guest_ip(&self, ip: &str) -> Result<()> { - fs::write(self.guest_ip_path(), ip).context("failed to write guest IP") - } - pub fn runtime_networks(&self) -> Vec { fs::read_to_string(self.runtime_networks_path()) .ok() diff --git a/dstack/vmm/src/config.rs b/dstack/vmm/src/config.rs index 330a2e440..d545f75f9 100644 --- a/dstack/vmm/src/config.rs +++ b/dstack/vmm/src/config.rs @@ -519,10 +519,6 @@ pub struct Networking { #[serde(default)] pub bridge: String, - /// Enable userspace port forwarding for bridge-mode VMs. - #[serde(default)] - pub forward_service_enabled: bool, - // ── MAC prefix ───────────────────────────────────────────────── /// Fixed MAC address prefix (0-3 colon-separated hex bytes, e.g. "02:ab:cd"). /// Remaining bytes are derived from the VM ID hash. diff --git a/dstack/vmm/src/main_service.rs b/dstack/vmm/src/main_service.rs index 932c4c17d..dbc35b383 100644 --- a/dstack/vmm/src/main_service.rs +++ b/dstack/vmm/src/main_service.rs @@ -10,12 +10,11 @@ use dstack_types::AppCompose; use dstack_vmm_rpc as rpc; use dstack_vmm_rpc::vmm_server::{VmmRpc, VmmServer}; use dstack_vmm_rpc::{ - AppId, ComposeHash as RpcComposeHash, DhcpLeaseRequest, GatewaySettings, GetInfoResponse, - GetMetaResponse, Id, ImageInfo as RpcImageInfo, ImageListResponse, KmsSettings, - ListGpusResponse, PublicKeyResponse, PullRegistryImageRequest, RegistryImageInfo, - RegistryImageListResponse, ReloadVmsResponse, ResizeVmRequest, ResourcesSettings, - StatusRequest, StatusResponse, SvListResponse, SvProcessInfo, UpdateVmRequest, VersionResponse, - VmConfiguration, + AppId, ComposeHash as RpcComposeHash, GatewaySettings, GetInfoResponse, GetMetaResponse, Id, + ImageInfo as RpcImageInfo, ImageListResponse, KmsSettings, ListGpusResponse, PublicKeyResponse, + PullRegistryImageRequest, RegistryImageInfo, RegistryImageListResponse, ReloadVmsResponse, + ResizeVmRequest, ResourcesSettings, StatusRequest, StatusResponse, SvListResponse, + SvProcessInfo, UpdateVmRequest, VersionResponse, VmConfiguration, }; use fs_err as fs; use or_panic::ResultOrPanic; @@ -225,7 +224,6 @@ fn networking_from_proto(proto: &rpc::NetworkingConfig) -> Result "bridge".to_string(), NetworkingMode::Custom => String::new(), }, - forward_service_enabled: default_networking.forward_service_enabled, default_bridge: default_networking.bridge.clone(), }), }) @@ -664,11 +658,6 @@ impl VmmRpc for RpcHandler { self.app.reload_vms_sync().await } - async fn report_dhcp_lease(self, request: DhcpLeaseRequest) -> Result<()> { - self.app.report_dhcp_lease(&request.mac, &request.ip).await; - Ok(()) - } - async fn sv_list(self) -> Result { use supervisor_client::supervisor::ProcessStatus; let list = self.app.supervisor.list().await?; @@ -929,27 +918,4 @@ mod tests { assert!(err.to_string().contains("custom networking mode")); } - - #[test] - fn multiple_bridges_are_rejected_when_builtin_forwarding_is_enabled() { - let mut cvm_config = test_cvm_config(); - cvm_config.networking.forward_service_enabled = true; - let mut request = test_vm_configuration(); - request.networks = vec![ - rpc::NetworkingConfig { - mode: "bridge".to_string(), - bridge_name: "lo".to_string(), - }, - rpc::NetworkingConfig { - mode: "bridge".to_string(), - bridge_name: "lo".to_string(), - }, - ]; - - let err = create_manifest_from_vm_config(request, &cvm_config).unwrap_err(); - - assert!(err - .to_string() - .contains("built-in port forwarding supports only one bridge")); - } } diff --git a/dstack/vmm/vmm.toml b/dstack/vmm/vmm.toml index 42ac81079..92c9f4b62 100644 --- a/dstack/vmm/vmm.toml +++ b/dstack/vmm/vmm.toml @@ -101,7 +101,6 @@ restrict = false # for mode = "bridge" # bridge = "virbr0" -forward_service_enabled = false [cvm.port_mapping] enabled = false