diff --git a/docs/libvirt-network-filter.md b/docs/libvirt-network-filter.md new file mode 100644 index 000000000..fc44513e0 --- /dev/null +++ b/docs/libvirt-network-filter.md @@ -0,0 +1,111 @@ +# Optional libvirt network filtering + +## Goal + +Allow bridge-backed VMs to opt into an existing libvirt `nwfilter` without +allowing libvirt to create or launch the QEMU domain. QEMU remains entirely +owned by `dstack-vmm`, so its command line and attestation inputs do not change +outside the explicitly selected network backend. + +The measurable acceptance criteria are: + +- `network_filter = "none"` preserves the existing QEMU `-netdev bridge` + behavior and does not require `netd` or libvirt. +- `network_filter = "libvirt"` creates the TAP and filter binding before QEMU + is submitted to Supervisor, and uses QEMU `-netdev tap`. +- A failed TAP or filter setup prevents QEMU from starting and rolls back all + interfaces prepared for that VM. +- Normal stop and removal delete the filter binding and TAP. +- One host `netd` can serve multiple VMM instances. Resource names include a + stable VMM instance namespace, VM ID, and NIC index. +- Development builds can run the VMM and `netd` directly, with explicit socket + and allowed-UID command-line options; systemd is not required. + +## Configuration + +Filtering is a VMM host policy, not a field accepted from a VM manifest: + +```toml +[cvm.network_filter] +mode = "none" # or "libvirt" +filter = "clean-traffic" +parameters = {} + +[netd] +socket = "/run/dstack/netd.sock" +allowed_uids = [] # empty means root only +libvirt_uri = "qemu:///system" +``` + +Each VMM instance also has an `instance_id`. It must be unique among VMMs that +share a host. If omitted, the VMM derives a stable namespace from its absolute +run directory. + +## Architecture + +`netd` is a host-level privilege broker. It accepts a small, bounded JSON +protocol over a Unix stream socket and authorizes clients with `SO_PEERCRED`. +A single process can serve multiple VMMs; a dedicated process can use another +socket for development or isolation. + +For libvirt mode, startup is: + +1. Derive the TAP name from instance namespace, VM ID, and NIC index. +2. Create the TAP for the configured QEMU UID and attach it to the bridge. +3. Create a libvirt nwfilter binding for the TAP. +4. Bring the TAP up and return success. +5. Start QEMU directly with `-netdev tap,script=no,downscript=no`. + +Teardown stops QEMU first, removes the binding, and deletes the TAP. Operations +are serialized by `netd`. The design intentionally does not add ownership +aliases; deployments must use unique instance IDs. + +`netd` invokes fixed absolute `ip` and `virsh` executables with separate +arguments. It never accepts a command, executable path, TAP name, or raw XML +from a client. Filter XML is generated internally with XML escaping and is +validated by libvirt. + +## Deployment modes + +Production should run one shared service. `netd` reads only the `[netd]` +section, so its root-owned configuration can be small and independent of every +VMM instance: + +```toml +# /etc/dstack/netd.toml +[netd] +socket = "/run/dstack/netd.sock" +allowed_uids = [991, 992] +libvirt_uri = "qemu:///system" +``` + +```ini +# /etc/systemd/system/dstack-netd.service +[Unit] +Description=dstack host networking service +After=libvirtd.service + +[Service] +ExecStart=/usr/bin/dstack-vmm --config /etc/dstack/netd.toml netd +Restart=on-failure + +[Install] +WantedBy=multi-user.target +``` + +All VMM instance configurations point to the same socket and use distinct +`cvm.instance_id` values. A dedicated netd uses a different socket. A +host-wide lock serializes mutations made by shared and dedicated netd +processes. + +Development mode is two ordinary commands: + +```bash +sudo dstack-vmm --config ./vmm.toml netd \ + --socket /run/dstack-dev/netd.sock --allow-uid "$(id -u)" +dstack-vmm --config ./vmm.toml \ + --netd-socket /run/dstack-dev/netd.sock +``` + +User networking and bridge networking with `mode = "none"` never connect to +`netd`. Libvirt mode fails closed if `netd` is unavailable. diff --git a/dstack/vmm/src/app.rs b/dstack/vmm/src/app.rs index 690b7c7af..5dec77716 100644 --- a/dstack/vmm/src/app.rs +++ b/dstack/vmm/src/app.rs @@ -2,7 +2,10 @@ // // SPDX-License-Identifier: Apache-2.0 -use crate::config::{Config, Networking, ProcessAnnotation, Protocol}; +use crate::{ + config::{Config, NetworkFilterMode, Networking, NetworkingMode, ProcessAnnotation, Protocol}, + netd::{self, InterfaceIdentity, PrepareRequest, Request as NetdRequest}, +}; use anyhow::{bail, Context, Result}; use bon::Builder; @@ -17,6 +20,7 @@ use dstack_vmm_rpc::{ use fs_err as fs; use guest_api::client::DefaultClient as GuestClient; use id_pool::IdPool; +use nix::unistd::User; use or_panic::ResultOrPanic; use ra_rpc::client::RaClient; use serde::{Deserialize, Serialize}; @@ -423,17 +427,30 @@ impl App { append_boot_separator(&work_dir.stdout_file()); append_boot_separator(&work_dir.stderr_file()); + let runtime_networks = resolved_networks(&vm_config.manifest, &self.config.cvm); let devices = self.try_allocate_gpus(&vm_config.manifest)?; let processes = vm_config.config_qemu(&work_dir, &self.config.cvm, &devices)?; - let runtime_networks = resolved_networks(&vm_config.manifest, &self.config.cvm); work_dir.set_runtime_networks(&runtime_networks)?; + if let Err(error) = self + .prepare_filtered_networks(&vm_config, &runtime_networks) + .await + { + let _ = work_dir.clear_runtime_networks(); + return Err(error); + } { let mut state = self.lock(); let vm_state = state.get_mut(id).context("VM not found")?; - vm_state.state.runtime_networks = runtime_networks; + vm_state.state.runtime_networks = runtime_networks.clone(); } for process in processes { if let Err(err) = self.supervisor.deploy(&process).await { + if let Err(cleanup_error) = self + .remove_filtered_networks(&vm_config.manifest.id, &runtime_networks) + .await + { + warn!(id, %cleanup_error, "failed to roll back filtered networking"); + } if let Err(clear_err) = work_dir.clear_runtime_networks() { warn!( id, @@ -465,6 +482,88 @@ impl App { pub async fn stop_vm(&self, id: &str) -> Result<()> { self.set_started(id, false)?; self.stop_vm_process(id).await?; + let networks = self.work_dir(id).runtime_networks(); + self.remove_filtered_networks(id, &networks).await?; + Ok(()) + } + + async fn prepare_filtered_networks( + &self, + vm: &VmConfig, + networks: &[Networking], + ) -> Result<()> { + if self.config.cvm.network_filter.mode == NetworkFilterMode::None { + return Ok(()); + } + let qemu_uid = if self.config.cvm.user.is_empty() { + unsafe { libc::geteuid() } + } else { + User::from_name(&self.config.cvm.user) + .context("failed to resolve QEMU user")? + .with_context(|| format!("QEMU user {} does not exist", self.config.cvm.user))? + .uid + .as_raw() + }; + let mut prepared = Vec::new(); + for (nic_index, network) in networks.iter().enumerate() { + if network.mode != NetworkingMode::Bridge { + continue; + } + let identity = InterfaceIdentity { + instance_id: self.config.cvm.instance_id.clone(), + vm_id: vm.manifest.id.clone(), + nic_index, + }; + let request = PrepareRequest { + identity: identity.clone(), + bridge: network.bridge.clone(), + mac: network::mac_address_for_vm_index( + &vm.manifest.id, + &network.mac_prefix_bytes(), + nic_index, + ), + qemu_uid, + filter: self.config.cvm.network_filter.filter.clone(), + parameters: self.config.cvm.network_filter.parameters.clone(), + }; + if let Err(error) = + netd::request(&self.config.netd.socket, &NetdRequest::Prepare(request)).await + { + for identity in prepared.into_iter().rev() { + let _ = + netd::request(&self.config.netd.socket, &NetdRequest::Remove { identity }) + .await; + } + return Err(error).context("failed to prepare libvirt-filtered networking"); + } + prepared.push(identity); + } + Ok(()) + } + + async fn remove_filtered_networks(&self, vm_id: &str, networks: &[Networking]) -> Result<()> { + if self.config.cvm.network_filter.mode == NetworkFilterMode::None { + return Ok(()); + } + let mut first_error = None; + for (nic_index, network) in networks.iter().enumerate().rev() { + if network.mode != NetworkingMode::Bridge { + continue; + } + let identity = InterfaceIdentity { + instance_id: self.config.cvm.instance_id.clone(), + vm_id: vm_id.to_string(), + nic_index, + }; + if let Err(error) = + netd::request(&self.config.netd.socket, &NetdRequest::Remove { identity }).await + { + first_error.get_or_insert(error); + } + } + if let Some(error) = first_error { + return Err(error).context("failed to remove libvirt-filtered networking"); + } Ok(()) } @@ -574,6 +673,11 @@ impl App { } } + let runtime_networks = self.work_dir(id).runtime_networks(); + if let Err(error) = self.remove_filtered_networks(id, &runtime_networks).await { + warn!(id, %error, "failed to remove filtered networking during VM removal"); + } + // Only delete the workdir for user-initiated removal or if .removing marker exists. // Orphaned supervisor processes without the marker keep their data intact. let vm_path = self.work_dir(id); diff --git a/dstack/vmm/src/app/qemu.rs b/dstack/vmm/src/app/qemu.rs index a68be81df..895378d44 100644 --- a/dstack/vmm/src/app/qemu.rs +++ b/dstack/vmm/src/app/qemu.rs @@ -14,7 +14,10 @@ use super::{ }; use crate::{ app::Manifest, - config::{CvmConfig, CvmPlatform, Networking, NetworkingMode, ProcessAnnotation}, + config::{ + CvmConfig, CvmPlatform, NetworkFilterMode, Networking, NetworkingMode, ProcessAnnotation, + }, + netd::{tap_name, InterfaceIdentity}, vm_launcher::{ChildCommand, LaunchSpec}, }; use anyhow::{bail, Context, Result}; @@ -600,7 +603,21 @@ impl QemuCommandBuilder<'_> { } NetworkingMode::Bridge => { tracing::info!("bridge networking: mac={mac} bridge={}", networking.bridge); - format!("bridge,id={net_id},br={}", networking.bridge) + match self.cfg.network_filter.mode { + NetworkFilterMode::None => { + format!("bridge,id={net_id},br={}", networking.bridge) + } + NetworkFilterMode::Libvirt => { + let tap = tap_name(&InterfaceIdentity { + instance_id: self.cfg.instance_id.clone(), + vm_id: self.vm.manifest.id.clone(), + nic_index: index, + }); + format!( + "tap,id={net_id},ifname={tap},script=no,downscript=no,vhost=off" + ) + } + } } NetworkingMode::Custom => { if !networking.netdev.contains(&format!("id={net_id}")) { @@ -976,7 +993,10 @@ mod tests { }; use crate::app::image::{Image, ImageInfo}; use crate::app::{needs_swtpm, GpuConfig, Manifest, PortMapping, VmVolume, VmWorkDir}; - use crate::config::{Config, CvmPlatform, Protocol, DEFAULT_CONFIG}; + use crate::config::{ + Config, CvmPlatform, NetworkFilterMode, NetworkingMode, Protocol, DEFAULT_CONFIG, + }; + use crate::netd::{tap_name, InterfaceIdentity}; use dstack_types::{KeyProviderKind, TeeVariant}; #[test] @@ -1173,6 +1193,42 @@ mod tests { .iter() .any(|arg| arg.contains("virtio-net-pci,netdev=net1"))); + for network in &mut prepared.networks { + network.mode = NetworkingMode::Bridge; + network.bridge = "br0".into(); + } + let process = QemuCommandBuilder { + vm: &vm, + cfg: &config.cvm, + gpus: &GpuConfig::default(), + prepared: &prepared, + } + .build() + .unwrap(); + assert!(process + .args + .iter() + .any(|arg| arg == "bridge,id=net0,br=br0")); + + config.cvm.instance_id = "vmm-a".into(); + config.cvm.network_filter.mode = NetworkFilterMode::Libvirt; + let process = QemuCommandBuilder { + vm: &vm, + cfg: &config.cvm, + gpus: &GpuConfig::default(), + prepared: &prepared, + } + .build() + .unwrap(); + let expected_tap = tap_name(&InterfaceIdentity { + instance_id: "vmm-a".into(), + vm_id: "vm-1".into(), + nic_index: 0, + }); + assert!(process.args.iter().any(|arg| { + arg == &format!("tap,id=net0,ifname={expected_tap},script=no,downscript=no,vhost=off") + })); + prepared.swtpm_socket = Some(PathBuf::from("/does-not-exist/vm-1/swtpm/swtpm.sock")); let process = QemuCommandBuilder { vm: &vm, diff --git a/dstack/vmm/src/config.rs b/dstack/vmm/src/config.rs index a31ad081f..1f9210c4e 100644 --- a/dstack/vmm/src/config.rs +++ b/dstack/vmm/src/config.rs @@ -2,7 +2,7 @@ // // SPDX-License-Identifier: Apache-2.0 -use std::{net::IpAddr, path::PathBuf, process::Command, str::FromStr}; +use std::{collections::BTreeMap, net::IpAddr, path::PathBuf, process::Command, str::FromStr}; use anyhow::{bail, Context, Result}; use load_config::load_config; @@ -314,6 +314,15 @@ pub struct CvmConfig { /// Networking configuration pub networking: Networking, + /// Optional host-side filtering for bridge interfaces. + #[serde(default)] + pub network_filter: NetworkFilterConfig, + + /// Stable namespace for TAP names when several VMMs share one host. + /// An empty value is derived from the absolute run directory. + #[serde(default)] + pub instance_id: String, + /// Host sharing mode. (9p, vhd, vvfat) pub host_share_mode: String, @@ -474,6 +483,10 @@ pub struct Config { /// CVM configuration pub cvm: CvmConfig, + + /// Privileged host networking service configuration. + #[serde(default)] + pub netd: NetdConfig, /// Gateway configuration pub gateway: GatewayConfig, @@ -490,6 +503,66 @@ pub struct Config { pub key_provider: KeyProviderConfig, } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize, Serialize)] +#[serde(rename_all = "lowercase")] +pub enum NetworkFilterMode { + #[default] + None, + Libvirt, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct NetworkFilterConfig { + #[serde(default)] + pub mode: NetworkFilterMode, + #[serde(default = "default_libvirt_filter")] + pub filter: String, + #[serde(default)] + pub parameters: BTreeMap, +} + +impl Default for NetworkFilterConfig { + fn default() -> Self { + Self { + mode: NetworkFilterMode::None, + filter: default_libvirt_filter(), + parameters: BTreeMap::new(), + } + } +} + +fn default_libvirt_filter() -> String { + "clean-traffic".to_string() +} + +#[derive(Debug, Clone, Deserialize)] +pub struct NetdConfig { + #[serde(default = "default_netd_socket")] + pub socket: PathBuf, + #[serde(default)] + pub allowed_uids: Vec, + #[serde(default = "default_libvirt_uri")] + pub libvirt_uri: String, +} + +impl Default for NetdConfig { + fn default() -> Self { + Self { + socket: default_netd_socket(), + allowed_uids: Vec::new(), + libvirt_uri: default_libvirt_uri(), + } + } +} + +fn default_netd_socket() -> PathBuf { + PathBuf::from("/run/dstack/netd.sock") +} + +fn default_libvirt_uri() -> String { + "qemu:///system".to_string() +} + #[derive(Debug, Default, Clone, Deserialize, Serialize)] pub struct ProcessAnnotation { #[serde(default)] diff --git a/dstack/vmm/src/main.rs b/dstack/vmm/src/main.rs index fb34bc362..f7e4049d4 100644 --- a/dstack/vmm/src/main.rs +++ b/dstack/vmm/src/main.rs @@ -7,7 +7,7 @@ use std::{path::Path, time::Duration}; use anyhow::{anyhow, Context, Result}; use app::App; use clap::{Args as ClapArgs, Parser, Subcommand}; -use config::Config; +use config::{Config, NetdConfig}; use dstack_api_auth::{Authenticator, HttpAuthConfig, HttpAuthFairing}; use guest_api_service::GuestApiHandler; use host_api_service::HostApiHandler; @@ -28,6 +28,7 @@ mod guest_api_service; mod host_api_service; mod main_routes; mod main_service; +mod netd; mod one_shot; mod openapi; mod vm_launcher; @@ -45,6 +46,9 @@ struct Args { /// Path to the configuration file #[arg(short, long)] config: Option, + /// Override the netd socket used by the VMM (useful without systemd). + #[arg(long, global = true)] + netd_socket: Option, /// Subcommand to run #[command(subcommand)] command: Option, @@ -57,11 +61,23 @@ enum Command { Serve, /// One-shot VM execution mode for debugging Run(RunArgs), + /// Run the privileged TAP and libvirt nwfilter broker. + Netd(NetdArgs), /// Internal per-VM QEMU/swtpm launcher. #[command(hide = true)] VmLauncher(VmLauncherArgs), } +#[derive(ClapArgs)] +struct NetdArgs { + /// Override the Unix socket configured in [netd]. + #[arg(long)] + socket: Option, + /// Authorize an additional client UID. May be repeated. + #[arg(long = "allow-uid")] + allow_uids: Vec, +} + #[derive(ClapArgs)] struct RunArgs { /// VM configuration file path @@ -177,7 +193,29 @@ async fn main() -> Result<()> { } let figment = config::load_config_figment(args.config.as_deref()); - let config = Config::extract_or_default(&figment)?.abs_path()?; + if let Some(Command::Netd(netd_args)) = &args.command { + let mut netd_config: NetdConfig = figment + .extract_inner("netd") + .context("failed to load [netd] configuration")?; + if let Some(socket) = args.netd_socket.as_deref() { + netd_config.socket = socket.into(); + } + if let Some(socket) = netd_args.socket.as_deref() { + netd_config.socket = socket.into(); + } + netd_config + .allowed_uids + .extend(netd_args.allow_uids.iter().copied()); + netd_config.allowed_uids.sort_unstable(); + netd_config.allowed_uids.dedup(); + return netd::serve(netd_config).await; + } + + let mut config = Config::extract_or_default(&figment)?.abs_path()?; + config.cvm.instance_id = netd::instance_id(&config.cvm.instance_id, config.run_path.as_path()); + if let Some(socket) = args.netd_socket.as_deref() { + config.netd.socket = socket.into(); + } // Validate host API configuration config @@ -188,6 +226,7 @@ async fn main() -> Result<()> { // Handle commands match args.command.unwrap_or_default() { Command::VmLauncher(_) => unreachable!("launcher mode handled before config loading"), + Command::Netd(_) => unreachable!("netd mode handled before server startup"), Command::Run(run_args) => { // One-shot VM execution mode return one_shot::run_one_shot( diff --git a/dstack/vmm/src/netd.rs b/dstack/vmm/src/netd.rs new file mode 100644 index 000000000..ec89141bd --- /dev/null +++ b/dstack/vmm/src/netd.rs @@ -0,0 +1,566 @@ +// SPDX-FileCopyrightText: © 2026 Phala Network +// +// SPDX-License-Identifier: Apache-2.0 + +//! Small privileged broker for TAP creation and libvirt nwfilter bindings. + +use std::{ + collections::BTreeMap, + fs::{File, OpenOptions, Permissions}, + io::Write as _, + os::{ + fd::AsRawFd, + unix::{fs::PermissionsExt, net::UnixStream as StdUnixStream}, + }, + path::Path, + process::{Command, Stdio}, + time::Duration, +}; + +use anyhow::{bail, Context, Result}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::{UnixListener, UnixStream}, + time::timeout, +}; +use tracing::{info, warn}; +use uuid::Uuid; + +use crate::config::NetdConfig; + +const MAX_MESSAGE_SIZE: u64 = 64 * 1024; +const IP_PATH: &str = "/usr/sbin/ip"; +const VIRSH_PATH: &str = "/usr/bin/virsh"; +const LOCK_PATH: &str = "/run/lock/dstack-netd.lock"; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct InterfaceIdentity { + pub instance_id: String, + pub vm_id: String, + pub nic_index: usize, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PrepareRequest { + #[serde(flatten)] + pub identity: InterfaceIdentity, + pub bridge: String, + pub mac: String, + pub qemu_uid: u32, + pub filter: String, + #[serde(default)] + pub parameters: BTreeMap, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "operation", rename_all = "snake_case")] +pub enum Request { + Prepare(PrepareRequest), + Remove { + #[serde(flatten)] + identity: InterfaceIdentity, + }, + Check { + #[serde(flatten)] + identity: InterfaceIdentity, + }, +} + +#[derive(Debug, Serialize, Deserialize)] +struct Response { + ok: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + tap: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + error: Option, +} + +pub fn tap_name(identity: &InterfaceIdentity) -> String { + let input = format!( + "{}\0{}\0{}", + identity.instance_id, identity.vm_id, identity.nic_index + ); + let digest = Sha256::digest(input.as_bytes()); + format!("dt{}", hex::encode(&digest[..6])) +} + +pub fn instance_id(configured: &str, run_path: &Path) -> String { + if !configured.trim().is_empty() { + return configured.trim().to_string(); + } + let digest = Sha256::digest(run_path.as_os_str().as_encoded_bytes()); + format!("path-{}", hex::encode(&digest[..8])) +} + +pub async fn request(socket: &Path, request: &Request) -> Result { + let operation = match request { + Request::Prepare(_) => "prepare", + Request::Remove { .. } => "remove", + Request::Check { .. } => "check", + }; + let exchange = async { + let mut stream = UnixStream::connect(socket) + .await + .with_context(|| format!("failed to connect to netd at {}", socket.display()))?; + let message = serde_json::to_vec(request)?; + if message.len() as u64 > MAX_MESSAGE_SIZE { + bail!("netd request is too large"); + } + stream.write_all(&message).await?; + stream.shutdown().await?; + let mut response = Vec::new(); + stream + .take(MAX_MESSAGE_SIZE + 1) + .read_to_end(&mut response) + .await?; + if response.len() as u64 > MAX_MESSAGE_SIZE { + bail!("netd response is too large"); + } + let response: Response = + serde_json::from_slice(&response).context("failed to decode netd response")?; + if !response.ok { + bail!( + "netd {operation} failed: {}", + response.error.as_deref().unwrap_or("unknown error") + ); + } + response.tap.context("netd response omitted TAP name") + }; + timeout(Duration::from_secs(30), exchange) + .await + .context("timed out waiting for netd")? +} + +pub async fn serve(config: NetdConfig) -> Result<()> { + if unsafe { libc::geteuid() } != 0 { + bail!("netd must run as root"); + } + require_executable(IP_PATH)?; + require_executable(VIRSH_PATH)?; + let listener = { + let _lock = OperationLock::acquire()?; + prepare_socket_path(&config.socket)?; + UnixListener::bind(&config.socket) + .with_context(|| format!("failed to bind netd socket {}", config.socket.display()))? + }; + // Access control is based on SO_PEERCRED rather than filesystem groups so + // one shared service can authorize VMM instances running under different + // UIDs and development mode needs no system group setup. + std::fs::set_permissions(&config.socket, Permissions::from_mode(0o666))?; + info!(socket = %config.socket.display(), "netd listening"); + loop { + let (mut stream, _) = listener.accept().await?; + let uid = peer_uid(&stream)?; + let allowed = uid == 0 || config.allowed_uids.contains(&uid); + let response = if allowed { + match read_request(&mut stream) + .await + .and_then(|request| handle_request(&config.libvirt_uri, request)) + { + Ok(tap) => Response { + ok: true, + tap: Some(tap), + error: None, + }, + Err(error) => { + warn!(%uid, %error, "netd request failed"); + Response { + ok: false, + tap: None, + error: Some(format!("{error:#}")), + } + } + } + } else { + warn!(%uid, "rejected unauthorized netd client"); + Response { + ok: false, + tap: None, + error: Some("caller UID is not authorized".into()), + } + }; + let encoded = serde_json::to_vec(&response)?; + stream.write_all(&encoded).await?; + stream.shutdown().await?; + } +} + +async fn read_request(stream: &mut UnixStream) -> Result { + let mut message = Vec::new(); + stream + .take(MAX_MESSAGE_SIZE + 1) + .read_to_end(&mut message) + .await?; + if message.len() as u64 > MAX_MESSAGE_SIZE { + bail!("request exceeds {MAX_MESSAGE_SIZE} bytes"); + } + serde_json::from_slice(&message).context("invalid netd request") +} + +fn handle_request(libvirt_uri: &str, request: Request) -> Result { + let _lock = OperationLock::acquire()?; + match request { + Request::Prepare(request) => prepare_interface(libvirt_uri, &request), + Request::Remove { identity } => { + validate_identity(&identity)?; + let tap = tap_name(&identity); + remove_interface(libvirt_uri, &tap)?; + Ok(tap) + } + Request::Check { identity } => { + validate_identity(&identity)?; + let tap = tap_name(&identity); + if !Path::new("/sys/class/net").join(&tap).exists() { + bail!("TAP {tap} does not exist"); + } + virsh(libvirt_uri, &["nwfilter-binding-dumpxml", &tap], None)?; + Ok(tap) + } + } +} + +struct OperationLock(File); + +impl OperationLock { + fn acquire() -> Result { + let file = OpenOptions::new() + .create(true) + .read(true) + .write(true) + .truncate(false) + .open(LOCK_PATH) + .with_context(|| format!("failed to open {LOCK_PATH}"))?; + // SAFETY: flock only acts on the valid file descriptor owned by file. + if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) } != 0 { + return Err(std::io::Error::last_os_error()).context("failed to lock netd operations"); + } + Ok(Self(file)) + } +} + +impl Drop for OperationLock { + fn drop(&mut self) { + // SAFETY: the file remains alive until after Drop returns. + unsafe { + libc::flock(self.0.as_raw_fd(), libc::LOCK_UN); + } + } +} + +fn prepare_interface(libvirt_uri: &str, request: &PrepareRequest) -> Result { + validate_prepare(request)?; + let tap = tap_name(&request.identity); + // A failed VMM start may leave a deterministic resource behind. Replacing + // it makes prepare idempotent without accepting a caller-selected TAP. + remove_interface(libvirt_uri, &tap)?; + + let uid = request.qemu_uid.to_string(); + ip(&["tuntap", "add", "dev", &tap, "mode", "tap", "user", &uid])?; + let result = (|| { + ip(&["link", "set", "dev", &tap, "master", &request.bridge])?; + let xml = binding_xml(request, &tap); + virsh( + libvirt_uri, + &["nwfilter-binding-create", "--validate", "/dev/stdin"], + Some(xml.as_bytes()), + )?; + ip(&["link", "set", "dev", &tap, "up"])?; + Ok(()) + })(); + if let Err(error) = result { + let _ = remove_interface(libvirt_uri, &tap); + return Err(error); + } + info!(%tap, bridge = %request.bridge, filter = %request.filter, "prepared filtered TAP"); + Ok(tap) +} + +fn remove_interface(libvirt_uri: &str, tap: &str) -> Result<()> { + if Path::new("/sys/class/net").join(tap).exists() { + let _ = ip(&["link", "set", "dev", tap, "down"]); + } + delete_binding(libvirt_uri, tap)?; + if Path::new("/sys/class/net").join(tap).exists() { + ip(&["link", "delete", "dev", tap])?; + info!(%tap, "removed filtered TAP"); + } + Ok(()) +} + +fn delete_binding(uri: &str, tap: &str) -> Result<()> { + let output = Command::new(VIRSH_PATH) + .args(["--connect", uri, "nwfilter-binding-delete", tap]) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .context("failed to execute virsh")?; + if output.status.success() { + return Ok(()); + } + let error = String::from_utf8_lossy(&output.stderr); + if error.contains("Network filter binding not found") { + return Ok(()); + } + bail!("virsh failed to delete binding {tap}: {}", error.trim()) +} + +fn binding_xml(request: &PrepareRequest, tap: &str) -> String { + let owner_uuid = stable_uuid(&request.identity); + let owner_name = format!( + "dstack:{}:{}:{}", + request.identity.instance_id, request.identity.vm_id, request.identity.nic_index + ); + let mut parameters = String::new(); + for (name, value) in &request.parameters { + parameters.push_str(&format!( + "", + xml_escape(name), + xml_escape(value) + )); + } + format!( + "{}{}\ + \ + {}", + xml_escape(&owner_name), + owner_uuid, + xml_escape(tap), + xml_escape(&request.mac), + xml_escape(&request.filter), + parameters + ) +} + +fn stable_uuid(identity: &InterfaceIdentity) -> Uuid { + let digest = Sha256::digest( + format!( + "{}\0{}\0{}", + identity.instance_id, identity.vm_id, identity.nic_index + ) + .as_bytes(), + ); + let mut bytes = [0_u8; 16]; + bytes.copy_from_slice(&digest[..16]); + bytes[6] = (bytes[6] & 0x0f) | 0x50; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + Uuid::from_bytes(bytes) +} + +fn validate_prepare(request: &PrepareRequest) -> Result<()> { + validate_identity(&request.identity)?; + validate_name("bridge", &request.bridge, 15, "_.-")?; + if !Path::new("/sys/class/net") + .join(&request.bridge) + .join("bridge") + .exists() + { + bail!("{} is not a host bridge", request.bridge); + } + validate_mac(&request.mac)?; + validate_name("filter", &request.filter, 128, "_.:-")?; + if request.parameters.len() > 64 { + bail!("too many nwfilter parameters"); + } + for (name, value) in &request.parameters { + validate_name("parameter name", name, 64, "_")?; + if value.len() > 512 || value.contains('\0') { + bail!("invalid nwfilter parameter value"); + } + } + Ok(()) +} + +fn validate_identity(identity: &InterfaceIdentity) -> Result<()> { + for (label, value) in [ + ("instance ID", identity.instance_id.as_str()), + ("VM ID", identity.vm_id.as_str()), + ] { + if value.is_empty() || value.len() > 128 || value.contains('\0') { + bail!("invalid {label}"); + } + } + if identity.nic_index > 255 { + bail!("NIC index is out of range"); + } + Ok(()) +} + +fn validate_name(label: &str, value: &str, max: usize, punctuation: &str) -> Result<()> { + if value.is_empty() + || value.len() > max + || !value + .chars() + .all(|ch| ch.is_ascii_alphanumeric() || punctuation.contains(ch)) + { + bail!("invalid {label}"); + } + Ok(()) +} + +fn validate_mac(mac: &str) -> Result<()> { + let bytes = mac + .split(':') + .map(|part| { + if part.len() != 2 { + bail!("invalid MAC address"); + } + u8::from_str_radix(part, 16).context("invalid MAC address") + }) + .collect::>>()?; + if bytes.len() != 6 || bytes[0] & 1 != 0 { + bail!("invalid unicast MAC address"); + } + Ok(()) +} + +fn xml_escape(value: &str) -> String { + value + .replace('&', "&") + .replace('<', "<") + .replace('>', ">") + .replace('\'', "'") + .replace('"', """) +} + +fn ip(args: &[&str]) -> Result<()> { + run_command(IP_PATH, args, None) +} + +fn virsh(uri: &str, args: &[&str], stdin: Option<&[u8]>) -> Result<()> { + let mut full_args = vec!["--connect", uri]; + full_args.extend_from_slice(args); + run_command(VIRSH_PATH, &full_args, stdin) +} + +fn run_command(program: &str, args: &[&str], stdin: Option<&[u8]>) -> Result<()> { + let mut child = Command::new(program) + .args(args) + .stdin(if stdin.is_some() { + Stdio::piped() + } else { + Stdio::null() + }) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .with_context(|| format!("failed to execute {program}"))?; + if let Some(input) = stdin { + child + .stdin + .take() + .context("missing command stdin")? + .write_all(input)?; + } + let output = child.wait_with_output()?; + if !output.status.success() { + let error = String::from_utf8_lossy(&output.stderr); + bail!("{} failed: {}", Path::new(program).display(), error.trim()); + } + Ok(()) +} + +fn require_executable(path: &str) -> Result<()> { + if !Path::new(path).is_file() { + bail!("required executable {path} does not exist"); + } + Ok(()) +} + +fn prepare_socket_path(socket: &Path) -> Result<()> { + let parent = socket.parent().context("netd socket has no parent")?; + std::fs::create_dir_all(parent) + .with_context(|| format!("failed to create {}", parent.display()))?; + if socket.exists() { + if StdUnixStream::connect(socket).is_ok() { + bail!("another netd is listening at {}", socket.display()); + } + std::fs::remove_file(socket) + .with_context(|| format!("failed to remove stale socket {}", socket.display()))?; + } + Ok(()) +} + +fn peer_uid(stream: &UnixStream) -> Result { + let mut credentials = libc::ucred { + pid: 0, + uid: 0, + gid: 0, + }; + let mut length = std::mem::size_of::() as libc::socklen_t; + // SAFETY: credentials and length point to valid writable storage of the + // exact size passed to getsockopt, and the stream owns a valid socket FD. + let result = unsafe { + libc::getsockopt( + stream.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_PEERCRED, + (&mut credentials as *mut libc::ucred).cast(), + &mut length, + ) + }; + if result != 0 { + return Err(std::io::Error::last_os_error()).context("failed to read peer credentials"); + } + Ok(credentials.uid) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn identity(instance: &str, vm: &str, nic_index: usize) -> InterfaceIdentity { + InterfaceIdentity { + instance_id: instance.into(), + vm_id: vm.into(), + nic_index, + } + } + + #[test] + fn tap_names_are_stable_bounded_and_namespaced() { + let first = tap_name(&identity("one", "vm", 0)); + assert_eq!(first, tap_name(&identity("one", "vm", 0))); + assert_ne!(first, tap_name(&identity("two", "vm", 0))); + assert_ne!(first, tap_name(&identity("one", "vm", 1))); + assert!(first.len() <= 15); + } + + #[test] + fn binding_xml_escapes_values() { + let request = PrepareRequest { + identity: identity("instance<&", "vm", 0), + bridge: "br0".into(), + mac: "02:00:00:00:00:01".into(), + qemu_uid: 1000, + filter: "clean-traffic".into(), + parameters: BTreeMap::from([("IP".into(), "10.0.0.2<&".into())]), + }; + let xml = binding_xml(&request, "dt123"); + assert!(xml.contains("instance<&")); + assert!(xml.contains("10.0.0.2<&")); + assert!(!xml.contains("instance<&")); + } + + #[test] + fn validation_rejects_injected_host_names() { + assert!(validate_name("bridge", "br0;id", 15, "_.-").is_err()); + assert!(validate_name("filter", "../../filter", 128, "_.:-").is_err()); + assert!(validate_mac("ff:ff:ff:ff:ff:ff").is_err()); + } + + #[test] + fn remove_protocol_keeps_identity_fields_flat() { + let request = Request::Remove { + identity: identity("instance", "vm", 2), + }; + let value = serde_json::to_value(request).unwrap(); + assert_eq!(value["operation"], "remove"); + assert_eq!(value["instance_id"], "instance"); + assert_eq!(value["vm_id"], "vm"); + assert_eq!(value["nic_index"], 2); + assert!(value.get("identity").is_none()); + } +} diff --git a/dstack/vmm/src/one_shot.rs b/dstack/vmm/src/one_shot.rs index ecbf9b83f..b208b4ba2 100644 --- a/dstack/vmm/src/one_shot.rs +++ b/dstack/vmm/src/one_shot.rs @@ -3,10 +3,10 @@ // SPDX-License-Identifier: Apache-2.0 use crate::app::{ - make_sys_config, simulator_config_for_manifest, sync_tee_simulator_config, Image, VmConfig, - VmWorkDir, + make_sys_config, resolved_networks, simulator_config_for_manifest, sync_tee_simulator_config, + Image, VmConfig, VmWorkDir, }; -use crate::config::Config; +use crate::config::{Config, NetworkFilterMode, NetworkingMode}; use crate::main_service; use anyhow::{Context, Result}; use fs_err as fs; @@ -294,6 +294,17 @@ Compose file content (first 200 chars): gateway_enabled: app_compose.gateway_enabled(), }; + if !dry_run + && config.cvm.network_filter.mode == NetworkFilterMode::Libvirt + && resolved_networks(&manifest, &config.cvm) + .iter() + .any(|network| network.mode == NetworkingMode::Bridge) + { + anyhow::bail!( + "one-shot execution does not manage libvirt-filtered TAP lifecycle; run the VMM server directly or use --dry-run" + ); + } + let process_configs = vm_builder_config .config_qemu(&workdir_path, &config.cvm, &gpus) .context("Failed to build QEMU configuration")?; diff --git a/dstack/vmm/vmm.toml b/dstack/vmm/vmm.toml index b405e5e9f..db7ad548d 100644 --- a/dstack/vmm/vmm.toml +++ b/dstack/vmm/vmm.toml @@ -44,6 +44,9 @@ max_allocable_memory_in_mb = 100_000 # MB qmp_socket = false # The user to run the VM as. If empty, the VM will be run as the current user. user = "" +# Unique namespace when multiple dstack-vmm instances share one host. When +# empty, a stable value is derived from run_path. +instance_id = "" use_mrconfigid = true # QEMU flags @@ -109,6 +112,20 @@ restrict = false # for mode = "bridge" # bridge = "virbr0" +# Optional filtering for bridge interfaces. "none" preserves the existing +# QEMU bridge-helper behavior and has no netd/libvirt dependency. +[cvm.network_filter] +mode = "none" +filter = "clean-traffic" +parameters = {} + +# Shared privileged networking service. Only used when network_filter.mode is +# "libvirt". An empty allowed_uids list permits root only. +[netd] +socket = "/run/dstack/netd.sock" +allowed_uids = [] +libvirt_uri = "qemu:///system" + [cvm.port_mapping] enabled = false address = "127.0.0.1"