use color_eyre::eyre;
use color_eyre::Result;
use std::path::Path;
use std::time::Duration;
pub const VSOCK_BRIDGE_MAX_RETRIES: u32 = 300;
#[cfg(unix)]
pub fn setup_vsock_ready_listener(env_hash: &str) -> Result<Option<std::os::unix::net::UnixListener>> {
let run_dir = &crate::models::dirs().epkg_run;
if let Ok(entries) = std::fs::read_dir(run_dir) {
for entry in entries.flatten() {
let name = entry.file_name().to_string_lossy().to_string();
if name.starts_with("vsock-") && name.ends_with(".sock") {
let _ = std::fs::remove_file(entry.path());
log::trace!("libkrun: cleaned up stale socket {}", name);
}
if name.starts_with("ready-") && name.ends_with(".sock") {
let _ = std::fs::remove_file(entry.path());
log::trace!("libkrun: cleaned up stale socket {}", name);
}
}
}
let ready_path = run_dir.join(format!("ready-{}.sock", env_hash));
let _ = std::fs::remove_file(&ready_path);
log::debug!("libkrun: creating ready listener on {}", ready_path.display());
let listener = std::os::unix::net::UnixListener::bind(&ready_path)
.map_err(|e| eyre::eyre!("Failed to bind ready socket {}: {}", ready_path.display(), e))?;
listener.set_nonblocking(true)
.map_err(|e| eyre::eyre!("Failed to set non-blocking on ready socket: {}", e))?;
Ok(Some(listener))
}
#[cfg(unix)]
pub fn wait_guest_ready_unix(
listener: &std::os::unix::net::UnixListener,
vm_start_failed_rx: Option<&std::sync::mpsc::Receiver<()>>,
) -> Result<()> {
use std::os::unix::io::AsRawFd;
use std::time::Instant;
let listener_fd = listener.as_raw_fd();
let start = Instant::now();
#[cfg(windows)]
let total_timeout = Duration::from_secs(300);
#[cfg(not(windows))]
let total_timeout = Duration::from_secs(30);
loop {
if let Some(ref failed_rx) = vm_start_failed_rx {
if failed_rx.try_recv().is_ok() {
return Err(eyre::eyre!("VM failed to start (krun_start_enter error)"));
}
}
let remaining = total_timeout.saturating_sub(start.elapsed());
if remaining.is_zero() {
log::error!("libkrun: timeout waiting for VM to become ready");
return Err(eyre::eyre!("Timeout waiting for VM to start"));
}
let remaining_ms = (remaining.as_millis().min(u32::MAX as u128) as u32) as i32;
let mut poll_fds = [libc::pollfd {
fd: listener_fd,
events: libc::POLLIN,
revents: 0,
}];
let poll_result = unsafe { libc::poll(poll_fds.as_mut_ptr(), 1, remaining_ms) };
match poll_result {
0 => {
}
n if n < 0 => {
log::error!("libkrun: poll error on ready socket");
return Err(eyre::eyre!("Poll error on ready socket"));
}
_ => {
let (stream, _addr) = listener
.accept()
.map_err(|e| eyre::eyre!("Failed to accept on ready socket: {}", e))?;
log::debug!("libkrun: guest connected to ready socket, guest is ready!");
drop(stream);
return Ok(());
}
}
}
}
#[cfg(unix)]
pub fn connect_vsock_bridge(sock_path: &Path, max_retries: u32) -> Result<std::os::unix::net::UnixStream> {
use std::os::unix::net::UnixStream;
let mut retry_count = 0;
let mut last_error = None;
while retry_count < max_retries {
match UnixStream::connect(sock_path) {
Ok(stream) => {
use std::os::unix::io::AsRawFd;
super::set_socket_buffer_size(stream.as_raw_fd());
return Ok(stream);
}
Err(e) => {
last_error = Some(e);
retry_count += 1;
if retry_count >= max_retries {
break;
}
std::thread::sleep(Duration::from_millis(5));
}
}
}
Err(eyre::eyre!(
"Failed to connect to Unix socket {} after {} retries: {}",
sock_path.display(),
max_retries,
last_error.unwrap_or_else(|| std::io::Error::new(std::io::ErrorKind::Other, "connection failed"))
))
}
#[cfg(windows)]
use std::os::windows::io::FromRawHandle;
#[cfg(windows)]
use windows::Win32::Foundation::{CloseHandle, HANDLE, INVALID_HANDLE_VALUE};
#[cfg(windows)]
use windows::Win32::Storage::FileSystem::{
CreateFileW, FILE_ATTRIBUTE_NORMAL, FILE_SHARE_READ, FILE_SHARE_WRITE,
OPEN_EXISTING, PIPE_ACCESS_DUPLEX,
};
#[cfg(windows)]
use windows::Win32::System::Pipes::{
ConnectNamedPipe, CreateNamedPipeW, WaitNamedPipeA, PIPE_READMODE_BYTE, PIPE_TYPE_BYTE,
PIPE_UNLIMITED_INSTANCES, PIPE_WAIT,
};
#[cfg(windows)]
use windows::Win32::Security::SECURITY_ATTRIBUTES;
#[cfg(windows)]
use windows::core::PCWSTR;
#[cfg(windows)]
pub(crate) fn pipe_name_from_sock_path(path: &Path) -> Result<String> {
path.file_stem()
.and_then(|s| s.to_str())
.map(|s| s.to_string())
.ok_or_else(|| eyre::eyre!("invalid vsock path (no file stem): {}", path.display()))
}
#[cfg(windows)]
fn to_wide_null(s: &str) -> Vec<u16> {
s.encode_utf16().chain(std::iter::once(0)).collect()
}
#[cfg(windows)]
pub struct WindowsReadyPipe {
handle: Option<HANDLE>,
}
#[cfg(windows)]
fn create_security_attributes() -> Result<(SECURITY_ATTRIBUTES, usize)> {
use windows::Win32::Security::{InitializeSecurityDescriptor, SetSecurityDescriptorDacl, PSECURITY_DESCRIPTOR};
use windows::Win32::Foundation::BOOL;
unsafe {
const SD_SIZE: usize = 256;
let sd_ptr = std::alloc::alloc(std::alloc::Layout::from_size_align(SD_SIZE, 8).unwrap());
if sd_ptr.is_null() {
return Err(eyre::eyre!("Failed to allocate security descriptor"));
}
if InitializeSecurityDescriptor(PSECURITY_DESCRIPTOR(sd_ptr as *mut core::ffi::c_void), 1).is_err() {
std::alloc::dealloc(sd_ptr, std::alloc::Layout::from_size_align(SD_SIZE, 8).unwrap());
return Err(eyre::eyre!("InitializeSecurityDescriptor failed"));
}
if SetSecurityDescriptorDacl(PSECURITY_DESCRIPTOR(sd_ptr as *mut core::ffi::c_void), BOOL(1), None, BOOL(0)).is_err() {
std::alloc::dealloc(sd_ptr, std::alloc::Layout::from_size_align(SD_SIZE, 8).unwrap());
return Err(eyre::eyre!("SetSecurityDescriptorDacl failed"));
}
Ok((SECURITY_ATTRIBUTES {
nLength: std::mem::size_of::<SECURITY_ATTRIBUTES>() as u32,
lpSecurityDescriptor: sd_ptr as *mut _,
bInheritHandle: BOOL(0),
}, SD_SIZE))
}
}
#[cfg(windows)]
impl Drop for WindowsReadyPipe {
fn drop(&mut self) {
if let Some(h) = self.handle.take() {
unsafe {
let _ = CloseHandle(h);
}
}
}
}
#[cfg(windows)]
impl WindowsReadyPipe {
fn take_handle(&mut self) -> Option<std::os::windows::io::RawHandle> {
self.handle.take().map(|h| h.0 as std::os::windows::io::RawHandle)
}
}
#[cfg(windows)]
pub fn setup_vsock_ready_listener(env_hash: &str) -> Result<Option<WindowsReadyPipe>> {
let run_dir = &crate::models::dirs().epkg_run;
let _ = std::fs::create_dir_all(run_dir);
let ready_path = run_dir.join(format!("ready-{}.sock", env_hash));
let pipe_name = pipe_name_from_sock_path(&ready_path)?;
let full = format!("\\\\.\\pipe\\{}", pipe_name);
let wide = to_wide_null(&full);
unsafe {
use windows::Win32::Foundation::BOOL;
let (security_attrs, sd_buffer_size) = create_security_attributes()
.unwrap_or_else(|e| {
crate::debug_epkg!("libkrun_bridge: warning - failed to create security attributes: {}, using default", e);
(SECURITY_ATTRIBUTES {
nLength: std::mem::size_of::<SECURITY_ATTRIBUTES>() as u32,
lpSecurityDescriptor: std::ptr::null_mut(),
bInheritHandle: BOOL(0),
}, 0usize)
});
let h = CreateNamedPipeW(
PCWSTR(wide.as_ptr()),
PIPE_ACCESS_DUPLEX,
PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
PIPE_UNLIMITED_INSTANCES,
4096,
4096,
0,
Some(&security_attrs),
);
if !security_attrs.lpSecurityDescriptor.is_null() {
std::alloc::dealloc(
security_attrs.lpSecurityDescriptor as *mut u8,
std::alloc::Layout::from_size_align(sd_buffer_size, 8).unwrap()
);
}
if h == INVALID_HANDLE_VALUE {
return Err(eyre::eyre!(
"CreateNamedPipeW failed: {}",
std::io::Error::last_os_error()
));
}
Ok(Some(WindowsReadyPipe { handle: Some(h) }))
}
}
#[cfg(windows)]
pub fn wait_guest_ready_windows(
pipe: &WindowsReadyPipe,
vm_start_failed_rx: Option<&std::sync::mpsc::Receiver<()>>,
) -> Result<()> {
use std::sync::mpsc;
use std::thread;
let handle_raw = pipe.handle.as_ref().map(|h| h.0 as usize).ok_or_else(|| eyre::eyre!("Pipe handle already taken"))?;
let (tx, rx) = mpsc::channel();
let jh = thread::spawn(move || {
let handle = HANDLE(handle_raw as *mut _);
let r = unsafe { ConnectNamedPipe(handle, None) };
let _ = tx.send(r);
});
let start = std::time::Instant::now();
#[cfg(windows)]
let timeout = Duration::from_secs(300);
#[cfg(not(windows))]
let timeout = Duration::from_secs(30);
let poll_interval = Duration::from_millis(10);
loop {
if let Some(ref failed_rx) = vm_start_failed_rx {
if failed_rx.try_recv().is_ok() {
return Err(eyre::eyre!("VM failed to start (krun_start_enter error)"));
}
}
match rx.recv_timeout(poll_interval) {
Ok(Ok(())) => {
log::debug!("libkrun: guest connected to ready pipe, guest is ready!");
let _ = jh.join();
return Ok(());
}
Ok(Err(e)) => {
let _ = jh.join();
return Err(eyre::eyre!("ConnectNamedPipe failed: {}", e));
}
Err(mpsc::RecvTimeoutError::Timeout) => {
}
Err(mpsc::RecvTimeoutError::Disconnected) => {
let _ = jh.join();
return Err(eyre::eyre!("Pipe thread disconnected unexpectedly"));
}
}
if start.elapsed() >= timeout {
log::error!("libkrun: timeout waiting for VM to become ready");
return Err(eyre::eyre!("Timeout waiting for VM to start"));
}
}
}
#[cfg(windows)]
pub fn connect_vsock_bridge(sock_path: &Path, max_retries: u32) -> Result<std::fs::File> {
let pipe_name = pipe_name_from_sock_path(sock_path)?;
let full = format!("\\\\.\\pipe\\{}", pipe_name);
crate::debug_epkg!("libkrun_bridge: connecting to named pipe: {}", full);
let c_path = std::ffi::CString::new(full.as_bytes())
.map_err(|_| eyre::eyre!("invalid pipe path"))?;
let mut retry_count = 0;
let mut last_error = None;
crate::debug_epkg!("libkrun_bridge: starting connection retry loop (max={})", max_retries);
while retry_count < max_retries {
unsafe {
crate::debug_epkg!("libkrun_bridge: attempt {} - waiting for named pipe...", retry_count);
if WaitNamedPipeA(
windows::core::PCSTR(c_path.as_ptr() as *const u8),
30_000,
)
.is_err()
{
last_error = Some(std::io::Error::last_os_error());
crate::debug_epkg!("libkrun_bridge: WaitNamedPipeA failed: {:?}", last_error);
retry_count += 1;
if retry_count >= max_retries {
break;
}
std::thread::sleep(Duration::from_millis(5));
continue;
}
crate::debug_epkg!("libkrun_bridge: named pipe is available, connecting...");
let access: u32 = 0x80000000u32 | 0x40000000u32;
let handle = CreateFileW(
PCWSTR(to_wide_null(&full).as_ptr()),
access,
FILE_SHARE_READ | FILE_SHARE_WRITE,
None,
OPEN_EXISTING,
FILE_ATTRIBUTE_NORMAL,
None,
);
match handle {
Ok(h) if h != INVALID_HANDLE_VALUE => {
crate::debug_epkg!("libkrun_bridge: successfully connected to named pipe");
let file = std::fs::File::from_raw_handle(h.0);
return Ok(file);
}
Ok(_) => {
crate::debug_epkg!("libkrun_bridge: CreateFileW returned INVALID_HANDLE_VALUE");
last_error = Some(std::io::Error::last_os_error());
}
Err(e) => {
crate::debug_epkg!("libkrun_bridge: CreateFileW failed: {}", e);
last_error = Some(std::io::Error::last_os_error());
}
}
}
retry_count += 1;
if retry_count < max_retries {
std::thread::sleep(Duration::from_millis(5));
}
}
crate::debug_epkg!("libkrun_bridge: failed to connect after {} retries: {:?}", max_retries, last_error);
Err(eyre::eyre!(
"Failed to connect to named pipe for {} after {} retries: {}",
sock_path.display(),
max_retries,
last_error.unwrap_or_else(|| std::io::Error::new(std::io::ErrorKind::Other, "connection failed"))
))
}
#[cfg(unix)]
#[allow(dead_code)]
pub fn setup_reverse_listener(sock_path: &Path) -> Result<std::os::unix::net::UnixListener> {
let _ = std::fs::remove_file(sock_path);
log::debug!("libkrun: creating reverse listener on {}", sock_path.display());
let listener = std::os::unix::net::UnixListener::bind(sock_path)
.map_err(|e| eyre::eyre!("Failed to bind reverse socket {}: {}", sock_path.display(), e))?;
listener.set_nonblocking(true)
.map_err(|e| eyre::eyre!("Failed to set non-blocking on reverse socket: {}", e))?;
Ok(listener)
}
#[cfg(unix)]
#[allow(dead_code)]
pub fn accept_reverse_connection(
listener: &std::os::unix::net::UnixListener,
vm_start_failed_rx: Option<&std::sync::mpsc::Receiver<()>>,
) -> Result<std::os::unix::net::UnixStream> {
use std::os::unix::io::AsRawFd;
use std::time::Instant;
let start = Instant::now();
let timeout = Duration::from_secs(30);
let listener_fd = listener.as_raw_fd();
loop {
if let Some(ref failed_rx) = vm_start_failed_rx {
if failed_rx.try_recv().is_ok() {
return Err(eyre::eyre!("VM failed to start (krun_start_enter error)"));
}
}
let remaining = timeout.saturating_sub(start.elapsed());
if remaining.is_zero() {
return Err(eyre::eyre!("Timeout waiting for Guest to connect (reverse mode)"));
}
let remaining_ms = (remaining.as_millis().min(u32::MAX as u128) as u32) as i32;
let mut poll_fds = [libc::pollfd {
fd: listener_fd,
events: libc::POLLIN,
revents: 0,
}];
let poll_result = unsafe { libc::poll(poll_fds.as_mut_ptr(), 1, remaining_ms) };
match poll_result {
0 => {
}
n if n < 0 => {
let errno = std::io::Error::last_os_error();
if errno.raw_os_error() == Some(libc::EINTR) {
continue;
}
return Err(eyre::eyre!("Poll error on reverse listener: {}", errno));
}
_ => {
match listener.accept() {
Ok((stream, _addr)) => {
log::debug!("libkrun: Guest connected to reverse listener");
stream.set_nonblocking(false)
.map_err(|e| eyre::eyre!("Failed to set blocking mode on reverse stream: {}", e))?;
return Ok(stream);
}
Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
}
Err(e) => {
return Err(eyre::eyre!("Failed to accept reverse connection: {}", e));
}
}
}
}
}
}
#[cfg(windows)]
pub fn setup_reverse_listener(sock_path: &Path) -> Result<WindowsReadyPipe> {
let pipe_name = pipe_name_from_sock_path(sock_path)?;
let full = format!("\\\\.\\pipe\\{}", pipe_name);
let wide = to_wide_null(&full);
unsafe {
use windows::Win32::Foundation::BOOL;
let (security_attrs, sd_buffer_size) = create_security_attributes()
.unwrap_or_else(|e| {
crate::debug_epkg!("libkrun_bridge: warning - failed to create security attributes: {}, using default", e);
(SECURITY_ATTRIBUTES {
nLength: std::mem::size_of::<SECURITY_ATTRIBUTES>() as u32,
lpSecurityDescriptor: std::ptr::null_mut(),
bInheritHandle: BOOL(0),
}, 0usize)
});
let h = CreateNamedPipeW(
PCWSTR(wide.as_ptr()),
PIPE_ACCESS_DUPLEX,
PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
PIPE_UNLIMITED_INSTANCES,
4096,
4096,
0,
Some(&security_attrs),
);
if !security_attrs.lpSecurityDescriptor.is_null() {
std::alloc::dealloc(
security_attrs.lpSecurityDescriptor as *mut u8,
std::alloc::Layout::from_size_align(sd_buffer_size, 8).unwrap()
);
}
if h == INVALID_HANDLE_VALUE {
return Err(eyre::eyre!(
"CreateNamedPipeW failed for reverse listener: {}",
std::io::Error::last_os_error()
));
}
log::debug!("libkrun: reverse listener created on pipe {}", full);
Ok(WindowsReadyPipe { handle: Some(h) })
}
}
#[cfg(windows)]
pub fn accept_reverse_connection(
mut pipe: WindowsReadyPipe,
vm_start_failed_rx: Option<&std::sync::mpsc::Receiver<()>>,
) -> Result<std::fs::File> {
use std::sync::mpsc;
use std::thread;
use std::time::Instant;
let handle_raw = pipe.take_handle()
.ok_or_else(|| eyre::eyre!("Pipe handle already taken"))? as usize;
let (tx, rx) = mpsc::channel();
crate::debug_epkg!("libkrun_bridge: spawning ConnectNamedPipe thread...");
let jh = thread::spawn(move || {
let handle = HANDLE(handle_raw as *mut _);
let r = unsafe { ConnectNamedPipe(handle, None) };
let _ = tx.send(r);
});
let start = Instant::now();
let timeout = Duration::from_secs(30);
let poll_interval = Duration::from_millis(10);
crate::debug_epkg!("libkrun_bridge: waiting for Guest connection or VM failure...");
loop {
if let Some(ref failed_rx) = vm_start_failed_rx {
match failed_rx.try_recv() {
Ok(_) => {
crate::debug_epkg!("libkrun_bridge: VM start failure detected!");
return Err(eyre::eyre!("VM failed to start (krun_start_enter error)"));
}
Err(mpsc::TryRecvError::Empty) => {
}
Err(mpsc::TryRecvError::Disconnected) => {
crate::debug_epkg!("libkrun_bridge: VM failure channel disconnected");
}
}
}
match rx.recv_timeout(poll_interval) {
Ok(Ok(())) => {
log::debug!("libkrun: Guest connected to reverse pipe");
crate::debug_epkg!("libkrun_bridge: Guest connected successfully!");
let _ = jh.join();
let file = std::fs::File::from(unsafe { std::os::windows::io::OwnedHandle::from_raw_handle(handle_raw as std::os::windows::io::RawHandle) });
return Ok(file);
}
Ok(Err(e)) => {
crate::debug_epkg!("libkrun_bridge: ConnectNamedPipe failed: {:?}", e);
let _ = jh.join();
return Err(eyre::eyre!("ConnectNamedPipe failed: {:?}", e));
}
Err(mpsc::RecvTimeoutError::Timeout) => {
}
Err(mpsc::RecvTimeoutError::Disconnected) => {
crate::debug_epkg!("libkrun_bridge: Pipe thread disconnected unexpectedly");
let _ = jh.join();
return Err(eyre::eyre!("Pipe thread disconnected unexpectedly"));
}
}
if start.elapsed() >= timeout {
crate::debug_epkg!("libkrun_bridge: Timeout waiting for Guest connection");
return Err(eyre::eyre!("Timeout waiting for Guest to connect (reverse mode)"));
}
}
}