Use vsock for communication with server.

This will allow implementing other network topologies.

Signed-off-by: Sasha Finkelstein <[email protected]>

Author: WhatAmISupposedToPutHere

crates/muvm/src/bin/muvm.rs

@@ -1,15 +1,15 @@
 use std::collections::HashMap;
-use std::env;
 use std::ffi::{c_char, CString};
 use std::io::Write;
 use std::os::fd::{IntoRawFd, OwnedFd};
 use std::path::Path;
 use std::process::ExitCode;
+use std::{env, fs};
 
 use anyhow::{anyhow, Context, Result};
 use krun_sys::{
-    krun_add_disk, krun_add_virtiofs2, krun_add_vsock_port, krun_create_ctx, krun_set_env,
-    krun_set_gpu_options2, krun_set_log_level, krun_set_passt_fd, krun_set_root,
+    krun_add_disk, krun_add_virtiofs2, krun_add_vsock_port, krun_add_vsock_port2, krun_create_ctx,
+    krun_set_env, krun_set_gpu_options2, krun_set_log_level, krun_set_passt_fd, krun_set_root,
     krun_set_vm_config, krun_set_workdir, krun_start_enter, VIRGLRENDERER_DRM,
     VIRGLRENDERER_THREAD_SYNC, VIRGLRENDERER_USE_ASYNC_FENCE_CB, VIRGLRENDERER_USE_EGL,
 };
@@ -22,7 +22,9 @@ use muvm::launch::{launch_or_lock, LaunchResult, DYNAMIC_PORT_RANGE};
 use muvm::monitor::spawn_monitor;
 use muvm::net::{connect_to_passt, start_passt};
 use muvm::types::MiB;
-use muvm::utils::launch::{GuestConfiguration, Launch};
+use muvm::utils::launch::{
+    GuestConfiguration, Launch, HIDPIPE_SOCKET, MUVM_GUEST_SOCKET, PULSE_SOCKET,
+};
 use nix::sys::sysinfo::sysinfo;
 use nix::unistd::User;
 use rustix::io::Errno;
@@ -72,8 +74,7 @@ fn main() -> Result<ExitCode> {
 
     let options = options().fallback_to_usage().run();
 
-    let (cookie, _lock_file, command, command_args, env) = match launch_or_lock(
-        options.server_port,
+    let (_lock_file, command, command_args, env) = match launch_or_lock(
         options.command,
         options.command_args,
         options.env,
@@ -87,12 +88,11 @@ fn main() -> Result<ExitCode> {
             return Ok(code);
         },
         LaunchResult::LockAcquired {
-            cookie,
             lock_file,
             command,
             command_args,
             env,
-        } => (cookie, lock_file, command, command_args, env),
+        } => (lock_file, command, command_args, env),
     };
 
     let mut env = prepare_env_vars(env).context("Failed to prepare environment variables")?;
@@ -264,9 +264,7 @@ fn main() -> Result<ExitCode> {
                 .context("Failed to connect to `passt`")?
                 .into()
         } else {
-            start_passt(options.server_port)
-                .context("Failed to start `passt`")?
-                .into()
+            start_passt().context("Failed to start `passt`")?.into()
         };
         // SAFETY: `passt_fd` is an `OwnedFd` and consumed to prevent closing on drop.
         // See https://doc.rust-lang.org/std/io/index.html#io-safety
@@ -287,7 +285,7 @@ fn main() -> Result<ExitCode> {
             )
             .context("Failed to process `pulse/native` path as it contains NUL character")?;
             // SAFETY: `pulse_path` is a pointer to a `CString` with long enough lifetime.
-            let err = unsafe { krun_add_vsock_port(ctx_id, 3333, pulse_path.as_ptr()) };
+            let err = unsafe { krun_add_vsock_port(ctx_id, PULSE_SOCKET, pulse_path.as_ptr()) };
             if err < 0 {
                 let err = Errno::from_raw_os_error(-err);
                 return Err(err).context("Failed to configure vsock for pulse socket");
@@ -304,7 +302,7 @@ fn main() -> Result<ExitCode> {
         .context("Failed to process `hidpipe` path as it contains NUL character")?;
 
         // SAFETY: `hidpipe_path` is a pointer to a `CString` with long enough lifetime.
-        let err = unsafe { krun_add_vsock_port(ctx_id, 3334, hidpipe_path.as_ptr()) };
+        let err = unsafe { krun_add_vsock_port(ctx_id, HIDPIPE_SOCKET, hidpipe_path.as_ptr()) };
         if err < 0 {
             let err = Errno::from_raw_os_error(-err);
             return Err(err).context("Failed to configure vsock for hidpipe socket");
@@ -328,6 +326,22 @@ fn main() -> Result<ExitCode> {
                 return Err(err).context("Failed to configure vsock for dynamic socket");
             }
         }
+
+        let server_path = Path::new(&run_path).join("krun/server");
+        _ = fs::remove_file(&server_path);
+        let server_path = CString::new(
+            server_path
+                .to_str()
+                .expect("server_path should not contain invalid UTF-8"),
+        )
+        .context("Failed to process `muvm-guest` path as it contains NUL characters")?;
+        // SAFETY: `server_path` is a pointer to a `CString` with long enough lifetime.
+        let err =
+            unsafe { krun_add_vsock_port2(ctx_id, MUVM_GUEST_SOCKET, server_path.as_ptr(), true) };
+        if err < 0 {
+            let err = Errno::from_raw_os_error(-err);
+            return Err(err).context("Failed to configure vsock for guest server socket");
+        }
     }
 
     let username = env::var("USER").context("Failed to get username from environment")?;
@@ -366,20 +380,17 @@ fn main() -> Result<ExitCode> {
     let display = env::var("DISPLAY").ok();
     let guest_config = GuestConfiguration {
         command: Launch {
-            cookie,
             command,
             command_args,
             env: HashMap::new(),
             vsock_port: 0,
             tty: false,
             privileged: false,
         },
-        server_port: options.server_port,
         username,
         uid: getuid().as_raw(),
         gid: getgid().as_raw(),
         host_display: display,
-        server_cookie: cookie,
         merged_rootfs: options.merged_rootfs,
     };
     let mut muvm_config_file = NamedTempFile::new()
@@ -449,7 +460,7 @@ fn main() -> Result<ExitCode> {
         }
     }
 
-    spawn_monitor(options.server_port, cookie);
+    spawn_monitor();
 
     {
         // Start and enter the microVM. Unless there is some error while creating the

crates/muvm/src/cli_options.rs

@@ -13,7 +13,6 @@ pub struct Options {
     pub mem: Option<MiB>,
     pub vram: Option<MiB>,
     pub passt_socket: Option<PathBuf>,
-    pub server_port: u32,
     pub fex_images: Vec<String>,
     pub merged_rootfs: bool,
     pub interactive: bool,
@@ -98,12 +97,6 @@ pub fn options() -> OptionParser<Options> {
         .help("Instead of starting passt, connect to passt socket at PATH")
         .argument("PATH")
         .optional();
-    let server_port = long("server-port")
-        .short('p')
-        .help("Set the port to be used in server mode")
-        .argument("SERVER_PORT")
-        .fallback(3334)
-        .display_fallback();
     let interactive = long("interactive")
         .short('i')
         .help("Attach to the command's stdin/out after starting it")
@@ -131,7 +124,6 @@ pub fn options() -> OptionParser<Options> {
         mem,
         vram,
         passt_socket,
-        server_port,
         fex_images,
         merged_rootfs,
         interactive,

crates/muvm/src/guest/bin/muvm-guest.rs

@@ -15,7 +15,7 @@ use muvm::guest::socket::setup_socket_proxy;
 use muvm::guest::user::setup_user;
 use muvm::guest::x11::setup_x11_forwarding;
 use muvm::guest::x11bridge::start_x11bridge;
-use muvm::utils::launch::GuestConfiguration;
+use muvm::utils::launch::{GuestConfiguration, PULSE_SOCKET};
 use nix::unistd::{Gid, Uid};
 use rustix::process::{getrlimit, setrlimit, Resource};
 
@@ -94,7 +94,7 @@ fn main() -> Result<()> {
     std::fs::create_dir(&pulse_path)
         .context("Failed to create `pulse` directory in `XDG_RUNTIME_DIR`")?;
     let pulse_path = pulse_path.join("native");
-    setup_socket_proxy(pulse_path, 3333)?;
+    setup_socket_proxy(pulse_path, PULSE_SOCKET)?;
 
     if let Some(host_display) = options.host_display {
         setup_x11_forwarding(run_path, &host_display)?;
@@ -108,13 +108,5 @@ fn main() -> Result<()> {
     });
 
     let rt = tokio::runtime::Runtime::new().unwrap();
-    rt.block_on(async {
-        server_main(
-            options.server_port,
-            options.server_cookie,
-            options.command.command,
-            options.command.command_args,
-        )
-        .await
-    })
+    rt.block_on(async { server_main(options.command.command, options.command.command_args).await })
 }

crates/muvm/src/guest/hidpipe.rs

@@ -3,6 +3,7 @@ use crate::hidpipe_common::{
     empty_input_event, struct_to_socket, AddDevice, ClientHello, FFErase, FFUpload, InputEvent,
     MessageType, RemoveDevice, ServerHello,
 };
+use crate::utils::launch::HIDPIPE_SOCKET;
 use input_linux::bitmask::BitmaskTrait;
 use input_linux::{
     AbsoluteAxis, AbsoluteInfo, Bitmask, EventKind, ForceFeedbackKind, InputProperty, Key, LedKind,
@@ -161,7 +162,7 @@ pub fn start_hidpipe(user_id: u32) {
         None,
     )
     .unwrap();
-    connect(sock_fd.as_raw_fd(), &VsockAddr::new(2, 3334)).unwrap();
+    connect(sock_fd.as_raw_fd(), &VsockAddr::new(2, HIDPIPE_SOCKET)).unwrap();
     let mut sock = UnixStream::from(sock_fd);
     let c_hello = ClientHello { version: 0 };
     let c_hello_data = unsafe {

crates/muvm/src/guest/server.rs

@@ -1,26 +1,41 @@
 use crate::guest::server_worker::{State, Worker};
+use crate::utils::launch::MUVM_GUEST_SOCKET;
 use anyhow::Result;
 use log::error;
+use nix::libc::VMADDR_CID_ANY;
+use nix::sys::socket::{
+    bind, listen, socket, AddressFamily, Backlog, SockFlag, SockType, VsockAddr,
+};
+use std::os::fd::AsRawFd;
+use std::os::unix::net::UnixListener as StdUnixListener;
 use std::os::unix::process::ExitStatusExt as _;
 use std::path::PathBuf;
-use tokio::net::TcpListener;
+use tokio::net::UnixListener;
 use tokio::process::Command;
 use tokio::sync::watch;
 use tokio_stream::wrappers::WatchStream;
 use tokio_stream::StreamExt as _;
-use uuid::Uuid;
 
-pub async fn server_main(
-    server_port: u32,
-    cookie: Uuid,
-    command: PathBuf,
-    command_args: Vec<String>,
-) -> Result<()> {
-    let listener = TcpListener::bind(format!("0.0.0.0:{server_port}")).await?;
+pub async fn server_main(command: PathBuf, command_args: Vec<String>) -> Result<()> {
+    let sock_fd = socket(
+        AddressFamily::Vsock,
+        SockType::Stream,
+        SockFlag::empty(),
+        None,
+    )
+    .unwrap();
+    bind(
+        sock_fd.as_raw_fd(),
+        &VsockAddr::new(VMADDR_CID_ANY, MUVM_GUEST_SOCKET),
+    )?;
+    listen(&sock_fd, Backlog::MAXCONN)?;
+    let std_listener = StdUnixListener::from(sock_fd);
+    std_listener.set_nonblocking(true)?;
+    let listener = UnixListener::from_std(std_listener)?;
     let (state_tx, state_rx) = watch::channel(State::new());
 
     let mut worker_handle = tokio::spawn(async move {
-        let mut worker = Worker::new(cookie, listener, state_tx);
+        let mut worker = Worker::new(listener, state_tx);
         worker.run().await;
     });
     let command_status = Command::new(&command).args(command_args).status();

crates/muvm/src/guest/server_worker.rs

@@ -5,7 +5,7 @@ use std::io::{Read, Write};
 use std::os::fd::{AsRawFd, OwnedFd};
 use std::os::unix::ffi::OsStringExt;
 use std::os::unix::fs::OpenOptionsExt;
-use std::os::unix::net::UnixStream;
+use std::os::unix::net::UnixStream as StdUnixStream;
 use std::os::unix::process::ExitStatusExt as _;
 use std::path::{Path, PathBuf};
 use std::process::{ExitStatus, Stdio};
@@ -21,13 +21,12 @@ use rustix::process::ioctl_tiocsctty;
 use rustix::pty::{ptsname, unlockpt};
 use rustix::termios::{tcsetwinsize, Winsize};
 use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufStream};
-use tokio::net::{TcpListener, TcpStream};
+use tokio::net::{UnixListener, UnixStream};
 use tokio::process::{Child, Command};
 use tokio::sync::watch;
 use tokio::task::{JoinError, JoinSet};
-use tokio_stream::wrappers::TcpListenerStream;
+use tokio_stream::wrappers::UnixListenerStream;
 use tokio_stream::StreamExt as _;
-use uuid::Uuid;
 
 use crate::guest::user;
 use crate::utils::launch::Launch;
@@ -45,8 +44,7 @@ pub enum ConnRequest {
 
 #[derive(Debug)]
 pub struct Worker {
-    cookie: Uuid,
-    listener_stream: TcpListenerStream,
+    listener_stream: UnixListenerStream,
     state_tx: watch::Sender<State>,
     child_set: JoinSet<(PathBuf, ChildResult, Option<OwnedFd>)>,
 }
@@ -60,10 +58,9 @@ pub struct State {
 type ChildResult = Result<ExitStatus, io::Error>;
 
 impl Worker {
-    pub fn new(cookie: Uuid, listener: TcpListener, state_tx: watch::Sender<State>) -> Self {
+    pub fn new(listener: UnixListener, state_tx: watch::Sender<State>) -> Self {
         Worker {
-            cookie,
-            listener_stream: TcpListenerStream::new(listener),
+            listener_stream: UnixListenerStream::new(listener),
             state_tx,
             child_set: JoinSet::new(),
         }
@@ -84,7 +81,7 @@ impl Worker {
                     };
                     let stream = BufStream::new(stream);
 
-                    match handle_connection(self.cookie, stream).await {
+                    match handle_connection(stream).await {
                         Ok(request) => match request {
                             ConnRequest::DropCaches => {},
                             ConnRequest::ExecuteCommand {command, mut child, stop_pipe } => {
@@ -180,7 +177,7 @@ impl Default for State {
     }
 }
 
-async fn read_request(stream: &mut BufStream<TcpStream>) -> Result<Launch> {
+async fn read_request(stream: &mut BufStream<UnixStream>) -> Result<Launch> {
     let mut buf = String::new();
     loop {
         if stream.read_line(&mut buf).await? == 0 {
@@ -193,14 +190,10 @@ async fn read_request(stream: &mut BufStream<TcpStream>) -> Result<Launch> {
     }
 }
 
-async fn handle_connection(
-    server_cookie: Uuid,
-    mut stream: BufStream<TcpStream>,
-) -> Result<ConnRequest> {
+async fn handle_connection(mut stream: BufStream<UnixStream>) -> Result<ConnRequest> {
     let mut envs: HashMap<String, String> = env::vars().collect();
 
     let Launch {
-        cookie,
         command,
         command_args,
         env,
@@ -209,13 +202,6 @@ async fn handle_connection(
         privileged,
     } = read_request(&mut stream).await?;
     debug!(command:?, command_args:?, env:?; "received launch request");
-    if cookie != server_cookie {
-        debug!("invalid cookie in launch request");
-        let msg = "Invalid cookie";
-        stream.write_all(msg.as_bytes()).await.ok();
-        stream.flush().await.ok();
-        return Err(anyhow!(msg));
-    }
 
     if command == Path::new("/muvmdropcaches") {
         // SAFETY: everything below should be async signal safe
@@ -370,7 +356,7 @@ fn run_io_guest(
         vsock_fd.as_raw_fd(),
         &VsockAddr::new(nix::libc::VMADDR_CID_HOST, vsock_port),
     )?;
-    let mut vsock = UnixStream::from(vsock_fd);
+    let mut vsock = StdUnixStream::from(vsock_fd);
     let epoll = Epoll::new(EpollCreateFlags::empty())?;
     epoll.add(&stdout, EpollEvent::new(EpollFlags::EPOLLIN, 1))?;
     if stderr.is_some() {

crates/muvm/src/guest/socket.rs

@@ -8,7 +8,7 @@ use anyhow::{Context, Result};
 use crate::utils::env::find_in_path;
 use crate::utils::stdio::make_stdout_stderr;
 
-pub fn setup_socket_proxy<P>(socket_path: P, port: u16) -> Result<()>
+pub fn setup_socket_proxy<P>(socket_path: P, port: u32) -> Result<()>
 where
     P: AsRef<Path>,
 {