diff --git a/litebox/src/broker/mod.rs b/litebox/src/broker/mod.rs index cbcdc47d7c..d1d4acdf98 100644 --- a/litebox/src/broker/mod.rs +++ b/litebox/src/broker/mod.rs @@ -535,6 +535,13 @@ pub(crate) trait BrokerControl: Send + Sync { flags: FileOpenFlags, ) -> core::result::Result; + /// Creates a vsock stream socket: see + /// [`CreateVsockSocketRequest`](litebox_broker_protocol::unix_socket::CreateVsockSocketRequest). + fn create_vsock_socket( + &self, + flags: FileOpenFlags, + ) -> core::result::Result; + fn create_unix_socket_pair( &self, socket_type: SocketType, @@ -555,12 +562,17 @@ pub(crate) trait BrokerControl: Send + Sync { backlog: u32, ) -> core::result::Result, BrokerControlError>; + /// Connects, or starts connecting to the host's vsock port: see + /// [`ConnectUnixSocketResponse`](litebox_broker_protocol::unix_socket::ConnectUnixSocketResponse). fn connect_unix_socket( &self, handle: ObjectHandle, address: &UnixSocketAddress, user: FileUser, - ) -> core::result::Result, BrokerControlError>; + ) -> core::result::Result< + core::result::Result, + BrokerControlError, + >; fn accept_unix_socket( &self, @@ -624,6 +636,29 @@ pub(crate) trait BrokerControl: Send + Sync { name: SocketOptionName, ) -> core::result::Result; + /// Takes why the socket's last connect failed, if no call reported it. + fn take_unix_socket_error( + &self, + handle: ObjectHandle, + ) -> core::result::Result, BrokerControlError>; + + /// Reports the outcome of the socket's vsock connect: see + /// [`CheckUnixSocketConnectRequest`](litebox_broker_protocol::unix_socket::CheckUnixSocketConnectRequest). + fn check_unix_socket_connect( + &self, + handle: ObjectHandle, + ) -> core::result::Result< + core::result::Result, + BrokerControlError, + >; + + /// Cancels the socket's vsock connect in progress: see + /// [`CancelUnixSocketConnectRequest`](litebox_broker_protocol::unix_socket::CancelUnixSocketConnectRequest). + fn cancel_unix_socket_connect( + &self, + handle: ObjectHandle, + ) -> core::result::Result, BrokerControlError>; + fn issue_shared_memory_id(&self) -> core::result::Result; fn create_futex_waiter(&self) -> core::result::Result; @@ -1607,6 +1642,13 @@ where self.request(|local| local.create_unix_socket(socket_type, flags)) } + fn create_vsock_socket( + &self, + flags: FileOpenFlags, + ) -> core::result::Result { + self.request(|local| local.create_vsock_socket(flags)) + } + fn create_unix_socket_pair( &self, socket_type: SocketType, @@ -1642,7 +1684,10 @@ where handle: ObjectHandle, address: &UnixSocketAddress, user: FileUser, - ) -> core::result::Result, BrokerControlError> { + ) -> core::result::Result< + core::result::Result, + BrokerControlError, + > { let Some(address) = address.encode() else { return Ok(Err(UnixSocketError::InvalidArgument)); }; @@ -1741,6 +1786,30 @@ where self.request(|local| local.get_unix_socket_option(handle, name)) } + fn take_unix_socket_error( + &self, + handle: ObjectHandle, + ) -> core::result::Result, BrokerControlError> { + self.request(|local| local.take_unix_socket_error(handle)) + } + + fn check_unix_socket_connect( + &self, + handle: ObjectHandle, + ) -> core::result::Result< + core::result::Result, + BrokerControlError, + > { + self.request(|local| local.check_unix_socket_connect(handle)) + } + + fn cancel_unix_socket_connect( + &self, + handle: ObjectHandle, + ) -> core::result::Result, BrokerControlError> { + self.request(|local| local.cancel_unix_socket_connect(handle)) + } + fn issue_shared_memory_id(&self) -> core::result::Result { self.request(BrokerLocal::issue_shared_memory_id) } diff --git a/litebox/src/unix_socket.rs b/litebox/src/unix_socket.rs index ed29385ebd..a813559aff 100644 --- a/litebox/src/unix_socket.rs +++ b/litebox/src/unix_socket.rs @@ -13,7 +13,9 @@ use litebox_broker_protocol::{ ObjectHandle, error::ErrorCode, fs::{FileMode, FileOpenFlags, FileStatusFlags, FileUser}, - socket::{ShutdownMode, SocketOptionName, SocketOptionValue, SocketType}, + socket::{ + ShutdownMode, SocketConnectionStatus, SocketOptionName, SocketOptionValue, SocketType, + }, unix_socket::{ UNIX_SOCKET_BUFFER_SIZE, UnixSocketAddress, UnixSocketError as ProtocolError, UnixSocketName, @@ -82,6 +84,22 @@ impl LiteBox { Ok(UnixSocket::new(self, broker, handle, socket_type)) } + /// Creates an unnamed, unconnected stream socket that the broker treats + /// as a vsock socket even before it has a name (see + /// [`CreateVsockSocketRequest`](litebox_broker_protocol::unix_socket::CreateVsockSocketRequest)). + /// + /// `flags` must be within [`FileOpenFlags::STATUS`]. + pub fn create_vsock_socket( + &self, + flags: FileOpenFlags, + ) -> Result, UnixSocketError> { + let broker = self.broker_control().ok_or(UnixSocketError::Io)?; + let handle = broker + .create_vsock_socket(flags) + .map_err(UnixSocketError::from)?; + Ok(UnixSocket::new(self, broker, handle, SocketType::Stream)) + } + /// Creates a pair of unnamed Unix sockets connected to each other. /// /// The arguments are as for [`Self::create_unix_socket`]. @@ -172,6 +190,21 @@ impl UnixSocket { /// A connection waits while the listener's backlog is full, unless /// `nonblock` is set or the socket is non-blocking, for at most the /// socket's send timeout. + /// + /// A stream socket's connect to the host's vsock port, or a connect to a + /// vsock address that finds one in progress, waits for it to complete, as + /// Linux does, with no timeout but the broker's, which fails the connect + /// ([`ProtocolError::TimedOut`]); it fails with the connect's error if it + /// fails, which, as in Linux, stays pending ([`Self::take_error`]) unless + /// the connect timed out. If it does not wait, it fails with + /// [`UnixSocketError::InProgress`], or, if it found one in progress, with + /// [`ProtocolError::AlreadyInProgress`]. As in Linux, an interrupted wait + /// cancels the connect, leaving the socket unconnected and still bound, + /// and fails with [`UnixSocketError::Interrupted`], unless the connect + /// completed meanwhile: then it reports its outcome. A wait that finds the + /// socket unconnected with no pending error, as when another call took + /// it, starts another attempt. A connect to any other address that finds + /// a vsock connect in progress, or the socket connected, fails at once. pub fn connect( &self, cx: &WaitContext<'_, Platform>, @@ -179,9 +212,71 @@ impl UnixSocket { user: FileUser, nonblock: bool, ) -> Result<(), UnixSocketError> { - self.wait(cx, nonblock, SocketOptionName::SendTimeout, || { - self.broker.connect_unix_socket(self.handle, address, user) - }) + let vsock = matches!(address, UnixSocketAddress::Vsock { .. }); + // The broker reports a vsock connect in progress whether or not the + // socket is non-blocking. + let nonblock = nonblock + || (vsock + && self + .broker + .get_status_flags(self.handle)? + .flags + .contains(FileOpenFlags::NONBLOCKING)); + let mut in_progress = None; + let timeout = (!vsock).then_some(SocketOptionName::SendTimeout); + let result = self.wait_for(cx, nonblock, timeout, || { + // A wait reads a vsock connect's outcome without taking its + // error, which a connect would. + let checked = if vsock && in_progress.is_some() { + self.broker.check_unix_socket_connect(self.handle) + } else { + Ok(Ok(SocketConnectionStatus::Unconnected)) + }; + let result = match checked { + Ok(Ok(SocketConnectionStatus::Unconnected)) => { + self.broker.connect_unix_socket(self.handle, address, user) + } + result => result, + }; + match result { + Ok(Ok(SocketConnectionStatus::Connected)) => Ok(()), + // Checked again as the socket's readiness changes. + Ok(Ok(_)) => { + in_progress.get_or_insert(InProgress::Started); + Err(TryOpError::TryAgain) + } + Ok(Err(ProtocolError::AlreadyInProgress)) if vsock => { + in_progress.get_or_insert(InProgress::Found); + Err(TryOpError::TryAgain) + } + // The connect in progress completed. + Ok(Err(ProtocolError::AlreadyConnected)) if vsock && in_progress.is_some() => { + Ok(()) + } + result => attempt(result).map(drop), + } + }); + match (result, in_progress) { + (Err(UnixSocketError::WouldBlock), Some(InProgress::Started)) => { + Err(UnixSocketError::InProgress) + } + (Err(UnixSocketError::WouldBlock), Some(InProgress::Found)) => { + Err(UnixSocketError::Socket(ProtocolError::AlreadyInProgress)) + } + (Err(UnixSocketError::WaitError(WaitError::Interrupted)), Some(_)) if vsock => { + match self.broker.cancel_unix_socket_connect(self.handle)? { + Ok(()) => Err(UnixSocketError::Interrupted), + Err(ProtocolError::AlreadyConnected) => Ok(()), + Err(error) => Err(UnixSocketError::Socket(error)), + } + } + (result, _) => result, + } + } + + /// Takes why the socket's last connect failed, if no call took it. + pub fn take_error(&self) -> Result, UnixSocketError> { + Ok(self.broker.take_unix_socket_error(self.handle)?) } /// Accepts a connection from a listening socket as a new socket of the @@ -341,17 +436,19 @@ impl UnixSocket { timeout: SocketOptionName, mut op: impl FnMut() -> Result, BrokerControlError>, ) -> Result { - let mut attempt = || match op() { - Ok(Ok(value)) => Ok(value), - Ok(Err(error)) => Err(TryOpError::Other(UnixSocketError::Socket(error))), - Err(BrokerControlError::Broker(ErrorCode::NonBlockingWouldBlock)) => { - Err(TryOpError::Other(UnixSocketError::WouldBlock)) - } - Err(error) => match BrokerObjectError::from(error) { - BrokerObjectError::WouldBlock => Err(TryOpError::TryAgain), - error => Err(TryOpError::Other(error.into())), - }, - }; + self.wait_for(cx, nonblock, Some(timeout), || attempt(op())) + } + + /// Runs `attempt` as [`Self::wait`] runs its operation, retrying while it + /// fails with [`TryOpError::TryAgain`]; with no `timeout`, it waits for as + /// long as it takes. + fn wait_for( + &self, + cx: &WaitContext<'_, Platform>, + nonblock: bool, + timeout: Option, + mut attempt: impl FnMut() -> Result>, + ) -> Result { match attempt() { Ok(value) => return Ok(value), Err(TryOpError::TryAgain) if !nonblock => {} @@ -359,10 +456,16 @@ impl UnixSocket { Err(TryOpError::Other(error)) => return Err(error), Err(TryOpError::WaitError(error)) => return Err(UnixSocketError::WaitError(error)), } - let timeout = match self.get_option(timeout)? { - SocketOptionValue::ReceiveTimeout(timeout) - | SocketOptionValue::SendTimeout(timeout) => timeout, - value => unreachable!("{value:?} is not a wait timeout"), + let timeout = match timeout + .map(|timeout| self.get_option(timeout)) + .transpose()? + { + Some( + SocketOptionValue::ReceiveTimeout(timeout) + | SocketOptionValue::SendTimeout(timeout), + ) => timeout, + None => None, + Some(value) => unreachable!("{value:?} is not a wait timeout"), }; self.pollee .wait( @@ -403,6 +506,32 @@ impl Drop for UnixSocket( + result: Result, BrokerControlError>, +) -> Result> { + match result { + Ok(Ok(value)) => Ok(value), + Ok(Err(error)) => Err(TryOpError::Other(UnixSocketError::Socket(error))), + Err(BrokerControlError::Broker(ErrorCode::NonBlockingWouldBlock)) => { + Err(TryOpError::Other(UnixSocketError::WouldBlock)) + } + Err(error) => match BrokerObjectError::from(error) { + BrokerObjectError::WouldBlock => Err(TryOpError::TryAgain), + error => Err(TryOpError::Other(error.into())), + }, + } +} + fn flatten( result: Result, BrokerControlError>, ) -> Result { @@ -447,6 +576,9 @@ pub mod errors { /// timeout expired. #[error("Unix socket operation would block")] WouldBlock, + /// A connect started, and did not wait for it to complete. + #[error("Unix socket connect is in progress")] + InProgress, #[error("wait error")] WaitError(WaitError), /// A wait bounded by the socket's timeout was interrupted, so the @@ -468,3 +600,204 @@ pub mod errors { Fault, } } + +#[cfg(test)] +mod tests { + use alloc::sync::Arc; + use alloc::vec::Vec; + use std::sync::Mutex; + + use litebox_broker_local::test_support::test_broker_local; + use litebox_broker_protocol::ObjectHandle; + use litebox_broker_protocol::fs::{FileAccessMode, FileOpenFlags, FileStatusFlags, FileUser}; + use litebox_broker_protocol::message::{ + BrokerOperation, BrokerRequest, BrokerResponse, BrokerResult, UnixSocketRequest, + UnixSocketResponse, + }; + use litebox_broker_protocol::readiness::ReadinessFlags; + use litebox_broker_protocol::shared_buffer::SHARED_BUFFER_POOL_SIZE; + use litebox_broker_protocol::socket::{SocketConnectionStatus, SocketType}; + use litebox_broker_protocol::unix_socket::{ + CheckUnixSocketConnectResponse, ConnectUnixSocketResponse, CreateUnixSocketResponse, + UnixSocketAddress, UnixSocketError as ProtocolError, VMADDR_CID_HOST, + }; + use litebox_broker_transport::channel::LocalCallChannel; + use litebox_broker_transport::shared_memory::HeapSharedMemory; + + use super::errors::UnixSocketError; + use crate::event::wait::{CheckForInterrupt, WaitState}; + use crate::platform::mock::MockPlatform; + + extern crate std; + + const SOCKET: ObjectHandle = ObjectHandle(7); + + /// A broker that answers connects with `connect`, and whose vsock + /// connects stay in progress until cancelled, when it answers `cancel`; + /// it records the Unix socket requests it serves. + struct PendingConnectChannel { + connect: Result, + cancel: Result<(), ProtocolError>, + requests: Arc>>, + } + + impl LocalCallChannel for PendingConnectChannel { + type Error = (); + + fn call(&self, request: BrokerRequest) -> Result { + let result = match request.operation { + BrokerOperation::UnixSocket(request) => { + self.requests.lock().unwrap().push(request.clone()); + BrokerResult::UnixSocket(match request { + UnixSocketRequest::Create(_) => { + UnixSocketResponse::Create(CreateUnixSocketResponse { handle: SOCKET }) + } + UnixSocketRequest::Connect(_) => match self.connect { + Ok(status) => { + UnixSocketResponse::Connect(ConnectUnixSocketResponse { status }) + } + Err(error) => UnixSocketResponse::Failed(error), + }, + UnixSocketRequest::CheckConnect(_) => { + UnixSocketResponse::CheckConnect(CheckUnixSocketConnectResponse { + status: SocketConnectionStatus::Connecting, + }) + } + UnixSocketRequest::CancelConnect(_) => match self.cancel { + Ok(()) => UnixSocketResponse::CancelConnect, + Err(error) => UnixSocketResponse::Failed(error), + }, + request => panic!("unexpected Unix socket request {request:?}"), + }) + } + BrokerOperation::GetStatusFlags(SOCKET) => { + BrokerResult::StatusFlags(FileStatusFlags { + access: FileAccessMode::ReadWrite, + flags: FileOpenFlags::NONE, + }) + } + BrokerOperation::CheckReadiness(SOCKET) => { + BrokerResult::Readiness(ReadinessFlags::default()) + } + BrokerOperation::CloseObject(SOCKET) => BrokerResult::ObjectClosed, + operation => panic!("unexpected broker operation {operation:?}"), + }; + Ok(BrokerResponse { + request_id: request.request_id, + result, + }) + } + } + + struct Interrupted; + + impl CheckForInterrupt for Interrupted { + fn check_for_interrupt(&self) -> bool { + true + } + } + + /// Interrupts a blocking connect to the host's vsock port while the + /// broker keeps it in progress, cancelling it with the result `cancel`; + /// returns the connect's result and the Unix socket requests it made. + fn interrupt_connect( + cancel: Result<(), ProtocolError>, + ) -> (Result<(), UnixSocketError>, Vec) { + connect_through( + Ok(SocketConnectionStatus::Connecting), + cancel, + &UnixSocketAddress::Vsock { + cid: VMADDR_CID_HOST, + port: 5000, + }, + ) + } + + /// Interrupts a blocking connect to `address`, which the broker answers + /// with `connect`, and, for a vsock connect in progress, cancels with + /// `cancel`; returns the connect's result and the Unix socket requests it + /// made. + fn connect_through( + connect: Result, + cancel: Result<(), ProtocolError>, + address: &UnixSocketAddress, + ) -> (Result<(), UnixSocketError>, Vec) { + let platform = MockPlatform::new(); + let requests = Arc::new(Mutex::new(Vec::new())); + let channel = PendingConnectChannel { + connect, + cancel, + requests: Arc::clone(&requests), + }; + let shared_memory = Arc::new(HeapSharedMemory::new(SHARED_BUFFER_POOL_SIZE)); + let litebox = crate::LiteBox::new_with_broker_local( + platform, + test_broker_local(channel, shared_memory), + ); + let socket = litebox + .create_unix_socket(SocketType::Stream, FileOpenFlags::NONE) + .unwrap(); + let wait = WaitState::new(platform); + let result = socket.connect( + &wait.context().with_check_for_interrupt(&Interrupted), + address, + FileUser::ROOT, + false, + ); + drop(socket); + let requests = requests.lock().unwrap().clone(); + (result, requests) + } + + #[test] + fn interrupted_vsock_connects_are_cancelled() { + let (result, requests) = interrupt_connect(Ok(())); + assert!(matches!(result, Err(UnixSocketError::Interrupted))); + // The wait checks the attempt rather than connecting again, has no + // timeout, so reads none, and cancels the attempt once. + assert!(matches!( + requests.as_slice(), + [ + UnixSocketRequest::Create(_), + UnixSocketRequest::Connect(_), + checks @ .., + UnixSocketRequest::CancelConnect(_), + ] if !checks.is_empty() + && checks + .iter() + .all(|request| matches!(request, UnixSocketRequest::CheckConnect(_))) + )); + + // A connect that completed meanwhile succeeds, and one that failed + // reports its error. + let (result, _) = interrupt_connect(Err(ProtocolError::AlreadyConnected)); + assert!(matches!(result, Ok(()))); + let (result, _) = interrupt_connect(Err(ProtocolError::ConnectionReset)); + assert!(matches!( + result, + Err(UnixSocketError::Socket(ProtocolError::ConnectionReset)) + )); + } + + #[test] + fn path_connects_fail_at_once_with_another_connects_progress() { + for error in [ + ProtocolError::AlreadyInProgress, + ProtocolError::AlreadyConnected, + ] { + let (result, requests) = connect_through( + Err(error), + Ok(()), + &UnixSocketAddress::Abstract(b"unix".into()), + ); + assert!(matches!( + result, + Err(UnixSocketError::Socket(found)) if found == error + )); + assert!(matches!( + requests.as_slice(), + [UnixSocketRequest::Create(_), UnixSocketRequest::Connect(_)] + )); + } + } +} diff --git a/litebox_broker_core/src/unix_socket.rs b/litebox_broker_core/src/unix_socket.rs index dbc3aaaeca..9e97a9fe2a 100644 --- a/litebox_broker_core/src/unix_socket.rs +++ b/litebox_broker_core/src/unix_socket.rs @@ -26,6 +26,58 @@ //! Readiness reaches every reference's registration from the moment the //! reference exists, since another process can change a socket through a //! connection without sharing a reference to it. +//! +//! Stream sockets may also hold vsock names, keyed apart from every path and +//! abstract name. A socket binds only a port at [`VMADDR_CID_ANY`], and no +//! socket sends to a vsock address. Only [`connect_from_vsock_host`] and +//! [`listen_for_vsock_host`], which the broker protocol does not reach, name +//! sockets at [`VMADDR_CID_HOST`], so a peer named at that CID is the host. +//! The cursor of ephemeral vsock ports lives in the registry of names. Its +//! seed is drawn without locks, since the random provider may block, and +//! installed under the registry's write lock unless another bind installed +//! one first; ports are taken only under that lock. +//! +//! The broker does not tie a socket to a family: any stream socket may bind +//! or connect either kind of address, and is a vsock socket if created as one +//! ([`create_vsock`]) or named by a vsock address. The shim keeps the +//! families apart: all of a broker's sockets belong to one guest, and only +//! the broker names a peer at [`VMADDR_CID_HOST`]. +//! +//! A stream socket connects to a vsock address only at the host's CID, and +//! only through the broker's host connector: the one socket +//! [`listen_for_vsock_host`] made listen, under a key no address names, for +//! the broker's own use. Without it, or at another CID, the address +//! is unreachable. The connect binds an unnamed socket to an ephemeral port, +//! queues on the connector a socket named as the host's port and connected to +//! it, and leaves it connecting: it has no peer name, neither sends nor +//! receives, and is neither readable nor writable. The connector's owner then +//! completes it ([`complete_vsock_host_connect`]): connected, or unconnected +//! again with a pending error, which [`take_error`] or the next connect +//! reports once, while the connector's socket's connection ends as if its +//! peer had closed. [`check_connect`] reports the outcome as a blocking +//! connect's wait does, leaving the error pending unless the connect timed +//! out. Data the connector's socket sends meanwhile is queued for the +//! connecting socket, which receives it, and reports it readable, only once +//! connected: every way a connect ends unconnected discards it. A connecting +//! socket that closes hangs up its peer as a connected one does, and one +//! whose peer closes first fails with +//! [`UnixSocketError::ConnectionReset`]. A connect cancelled +//! ([`cancel_vsock_connect`]) leaves the socket unconnected, still bound, and +//! hangs up its peer as a close does. A connecting socket and its peer hold +//! each other as a connected pair does, and the connect locks the socket's +//! and the connector's states as a connect locks a listener's. +//! +//! The host sees a guest socket's port as the connection's source port. +//! Guests bind any port, those below 1024 included, with no privilege check, +//! so the host must not trust a guest's source port. +//! +//! A vsock socket's own shutdown differs from a Unix socket's, as in Linux: +//! once it shuts down receiving, receives report the end of data at once, +//! even with data queued, and once it shuts down either direction, it has no +//! peer name. Its peer's shutdown does not affect it so. A vsock socket that +//! is not connected does not shut down, and, unlike an unconnected Unix +//! stream socket, is not hung up: it is writable, as in Linux. A socket +//! created as a vsock socket is one even before it has a name. use alloc::collections::VecDeque; use alloc::sync::{Arc, Weak}; @@ -39,11 +91,12 @@ use litebox_broker_protocol::fs::{ }; use litebox_broker_protocol::readiness::ReadinessFlags; use litebox_broker_protocol::socket::{ - ShutdownMode, SocketOptionName, SocketOptionValue, SocketType, + ShutdownMode, SocketConnectionStatus, SocketOptionName, SocketOptionValue, SocketType, }; use litebox_broker_protocol::unix_socket::{ MAX_UNIX_SOCKET_BACKLOG, MAX_UNIX_SOCKET_TRANSFER_SIZE, UNIX_SOCKET_BUFFER_SIZE, - UnixSocketAddress, UnixSocketError, UnixSocketName, + UnixSocketAddress, UnixSocketError, UnixSocketName, VMADDR_CID_ANY, VMADDR_CID_HOST, + VMADDR_PORT_ANY, }; use litebox_platform::buffer::{ReadBuffer, WriteBuffer}; use spin::{Mutex, MutexGuard, rwlock::RwLock}; @@ -70,6 +123,16 @@ const DATAGRAM_OVERHEAD: usize = 256; /// which holds no reference until accepted. const CONNECTION_OVERHEAD: usize = 256; +/// The lowest ephemeral vsock port, above the ports Linux reserves. +const FIRST_EPHEMERAL_VSOCK_PORT: u32 = 1024; + +/// Number of ephemeral vsock ports: every port from +/// [`FIRST_EPHEMERAL_VSOCK_PORT`] but [`VMADDR_PORT_ANY`]. +const EPHEMERAL_VSOCK_PORTS: u32 = VMADDR_PORT_ANY - FIRST_EPHEMERAL_VSOCK_PORT; + +/// Ephemeral vsock ports one bind tries before it fails. +const MAX_EPHEMERAL_VSOCK_PORT_ATTEMPTS: u32 = 64; + /// Result of receiving from a Unix socket. #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct Received { @@ -92,11 +155,32 @@ pub fn create( socket_type: SocketType, flags: FileOpenFlags, readiness_sink: &Arc, +) -> Result { + create_socket(process, socket_type, flags, false, readiness_sink) +} + +/// Creates an unconnected vsock stream socket: as [`create`] does, but, like +/// a socket named by a vsock address, it is not hung up while unconnected, +/// even before it has a name (see the module docs). +pub fn create_vsock( + process: &BrokerProcess, + flags: FileOpenFlags, + readiness_sink: &Arc, +) -> Result { + create_socket(process, SocketType::Stream, flags, true, readiness_sink) +} + +fn create_socket( + process: &BrokerProcess, + socket_type: SocketType, + flags: FileOpenFlags, + vsock: bool, + readiness_sink: &Arc, ) -> Result { let (rights, status) = creation_rights(process, flags)?; let reference = process.reserve_object_reference(rights)?; let registration = ReadinessRegistration::new(reference.handle(), Arc::clone(readiness_sink)); - let object = UnixSocketObject::new(process, socket_type, status)?; + let object = UnixSocketObject::new(process, socket_type, status, vsock)?; object.watch(®istration)?; reference.commit_with_readiness(ObjectEntry::UnixSocket(object), Some(registration)) } @@ -119,8 +203,8 @@ pub fn create_pair( ReadinessRegistration::new(first_reference.handle(), Arc::clone(readiness_sink)); let second_registration = ReadinessRegistration::new(second_reference.handle(), Arc::clone(readiness_sink)); - let first = UnixSocketObject::new(process, socket_type, status)?; - let second = UnixSocketObject::new(process, socket_type, status)?; + let first = UnixSocketObject::new(process, socket_type, status, false)?; + let second = UnixSocketObject::new(process, socket_type, status, false)?; for (socket, peer) in [ (&first.socket, &second.socket), (&second.socket, &first.socket), @@ -149,6 +233,11 @@ pub fn create_pair( /// /// Binding to a path creates a file there with `mode` as `user`, which fails /// with [`UnixSocketError::AddressInUse`] if the path already exists. +/// +/// Only a stream socket binds a vsock address, whose CID must be +/// [`VMADDR_CID_ANY`]. Binding [`VMADDR_PORT_ANY`] binds an unused port above +/// 1023, or fails with [`UnixSocketError::AddressNotAvailable`] if the ports +/// it tries are in use. pub fn bind( process: &BrokerProcess, handle: ObjectHandle, @@ -158,19 +247,45 @@ pub fn bind( ) -> Result> { let lease = Lease::new(process.authorized_object(handle, ObjectRights::WRITE)?)?; let socket = &lease.socket; + // A socket never loses its name, so one unnamed at the bind below was + // unnamed here, and the cursor is seeded before it is taken. + if matches!( + address, + UnixSocketAddress::Vsock { + cid: VMADDR_CID_ANY, + port: VMADDR_PORT_ANY, + } + ) && socket.kind == SocketType::Stream + && socket.state.lock().name == UnixSocketName::Unnamed + { + socket.sockets.seed_ephemeral_vsock_ports(process); + } let path = { let mut state = socket.state.lock(); if state.binding || state.name != UnixSocketName::Unnamed { return Ok(Err(UnixSocketError::InvalidArgument)); } match address { - UnixSocketAddress::Abstract(_) => { + UnixSocketAddress::Abstract(bytes) => { return socket.register_name( &mut state, - NameKey::of(address, socket.kind), + NameKey::Abstract(socket.kind, bytes.clone()), address.name(), ); } + &UnixSocketAddress::Vsock { cid, port } => { + if socket.kind != SocketType::Stream { + return Ok(Err(UnixSocketError::InvalidArgument)); + } + if cid != VMADDR_CID_ANY { + return Ok(Err(UnixSocketError::AddressNotAvailable)); + } + return if port == VMADDR_PORT_ANY { + socket.register_ephemeral_vsock_port(&mut state) + } else { + socket.register_name(&mut state, NameKey::Vsock(port), address.name()) + }; + } UnixSocketAddress::Path { path, .. } => { state.binding = true; path @@ -235,7 +350,9 @@ pub fn listen( }; } Connection::Listening { limit: current, .. } => *current = limit, - Connection::Connected { .. } => return Ok(Err(UnixSocketError::InvalidArgument)), + Connection::Connecting { .. } | Connection::Connected { .. } => { + return Ok(Err(UnixSocketError::InvalidArgument)); + } } // Sockets waiting for room in the backlog retry. socket.incoming.lock().wake() @@ -249,15 +366,32 @@ pub fn listen( /// /// A stream or sequenced-packet socket queues a new connection on the /// listening socket there, failing with [`BrokerError::WouldBlock`] while its -/// backlog is full. A datagram socket sets its default destination. +/// backlog is full. A datagram socket sets its default destination. Either is +/// then [`SocketConnectionStatus::Connected`]. +/// +/// A stream socket's connect to the host's vsock port is +/// [`SocketConnectionStatus::Connecting`] until it completes (see the module +/// docs); it fails with [`BrokerError::ResourceExhausted`] while the host +/// connector's backlog is full. Any other vsock address is +/// [`UnixSocketError::Unreachable`]. As in Linux, a stream or +/// sequenced-packet socket's connect to a vsock address fails first by its +/// state: with [`UnixSocketError::InvalidArgument`] if it listens or is +/// connected and shut down, with [`UnixSocketError::AlreadyConnected`] if it +/// is otherwise connected, with [`UnixSocketError::AlreadyInProgress`] if it +/// is connecting, and with its pending error, which it takes, if it has one. pub fn connect( process: &BrokerProcess, handle: ObjectHandle, address: &UnixSocketAddress, user: FileUser, -) -> Result> { +) -> Result> { let lease = Lease::new(process.authorized_object(handle, ObjectRights::WRITE)?)?; let socket = &lease.socket; + if let &UnixSocketAddress::Vsock { cid, port } = address + && socket.is_connection_oriented() + { + return socket.connect_to_vsock_host(process, cid, port); + } // The node stays open until the connect finishes, so its identity cannot // be reused meanwhile. let (key, _node) = match lookup(process, address, user, socket.kind)? { @@ -268,13 +402,252 @@ pub fn connect( Ok(target) => target, Err(error) => return Ok(Err(error)), }; - match socket.kind { + let connected = match socket.kind { SocketType::Stream | SocketType::SeqPacket => { - socket.connect_to_listener(&target, lease.nonblocking) + socket.connect_to_listener(&target, lease.nonblocking)? } - SocketType::Datagram => Ok(socket.connect_datagram(&target)), - _ => Err(BrokerError::Internal), + SocketType::Datagram => socket.connect_datagram(&target), + _ => return Err(BrokerError::Internal), + }; + Ok(connected.map(|()| SocketConnectionStatus::Connected)) +} + +/// Names the unnamed, unconnected stream socket `handle` as the VM host's +/// vsock port `host_port`, and connects it, without waiting, to the socket +/// listening on vsock port `guest_port`. +/// +/// This is the broker's own entry point for the host's connections, which no +/// broker request reaches, so only the host's sockets bear its CID. Fails with +/// [`UnixSocketError::ConnectionRefused`] if no socket listens on the port, +/// and with [`BrokerError::NonBlockingWouldBlock`] while its backlog is full. +/// The socket keeps its name if the connect fails. +pub fn connect_from_vsock_host( + process: &BrokerProcess, + handle: ObjectHandle, + guest_port: u32, + host_port: u32, +) -> Result> { + let lease = Lease::new(process.authorized_object(handle, ObjectRights::WRITE)?)?; + let socket = &lease.socket; + if socket.kind != SocketType::Stream { + return Ok(Err(UnixSocketError::WrongType)); + } + { + let mut state = socket.state.lock(); + if state.binding + || state.name != UnixSocketName::Unnamed + || !matches!(state.connection, Connection::None) + { + return Ok(Err(UnixSocketError::InvalidArgument)); + } + // Host ports name no key, so no guest can bind or connect to them. + state.name = UnixSocketName::Vsock { + cid: VMADDR_CID_HOST, + port: host_port, + }; } + let target = match socket + .sockets + .find(&NameKey::Vsock(guest_port), SocketType::Stream) + { + Ok(target) => target, + Err(error) => return Ok(Err(error)), + }; + socket.connect_to_listener(&target, true) +} + +/// Makes the unnamed, unconnected stream socket `handle` the broker's host +/// connector (see the module docs), with up to `backlog` connects waiting +/// beyond the first. +/// +/// This is the broker's own entry point, which no broker request reaches. +/// Each socket the connector accepts is named as the host's vsock port its +/// peer connects to, and the peer is named as the guest's port it connects +/// from; its owner completes the connect with +/// [`complete_vsock_host_connect`]. Fails with +/// [`UnixSocketError::AddressInUse`] if the broker has a connector. +pub fn listen_for_vsock_host( + process: &BrokerProcess, + handle: ObjectHandle, + backlog: u32, +) -> Result> { + let lease = Lease::new(process.authorized_object(handle, ObjectRights::WRITE)?)?; + let socket = &lease.socket; + if socket.kind != SocketType::Stream { + return Ok(Err(UnixSocketError::WrongType)); + } + let mut state = socket.state.lock(); + if state.binding + || state.name != UnixSocketName::Unnamed + || !matches!(state.connection, Connection::None) + { + return Ok(Err(UnixSocketError::InvalidArgument)); + } + // Its name, which no key the guests' addresses name holds, keeps it from + // binding. + let name = UnixSocketName::Vsock { + cid: VMADDR_CID_HOST, + port: VMADDR_PORT_ANY, + }; + if let Err(error) = socket.register_name(&mut state, NameKey::VsockHost, name)? { + return Ok(Err(error)); + } + state.connection = Connection::Listening { + backlog: VecDeque::new(), + limit: backlog.min(MAX_UNIX_SOCKET_BACKLOG) as usize, + }; + Ok(Ok(())) +} + +/// Completes the connect of the socket that `handle`, which the host +/// connector accepted, is connected to: it is connected if `result` is `Ok`, +/// receiving the data `handle` sent meanwhile. Else it is unconnected with +/// the pending error `result`, that data is discarded, and `handle`'s +/// connection ends as if its peer had closed: it no longer reaches that +/// socket or sends, and receives the end of data. +/// +/// This is the broker's own entry point, which no broker request reaches. +/// Fails with [`UnixSocketError::NotConnected`], changing nothing, if that +/// socket closed, or `handle` is not connecting it. +pub fn complete_vsock_host_connect( + process: &BrokerProcess, + handle: ObjectHandle, + result: UnixSocketResult<()>, +) -> Result> { + let lease = Lease::new(process.authorized_object(handle, ObjectRights::WRITE)?)?; + let host = &lease.socket; + let guest = match &host.state.lock().connection { + Connection::Connected { peer, .. } => peer.upgrade(), + _ => None, + }; + let Some(guest) = guest else { + return Ok(Err(UnixSocketError::NotConnected)); + }; + let (guest_wakeups, host_wakeups) = { + let (mut host_state, mut guest_state) = lock_both(host, &guest); + let peer_name = match &guest_state.connection { + Connection::Connecting { peer, peer_name } if refers_to(peer, host) => { + peer_name.clone() + } + _ => return Ok(Err(UnixSocketError::NotConnected)), + }; + match result { + Ok(()) => { + guest_state.connection = Connection::Connected { + peer: Arc::downgrade(host), + peer_name, + }; + (Wakeups::default(), Wakeups::default()) + } + Err(error) => ( + guest.end_connect(&mut guest_state, Some(error)), + host.end_connection(&mut host_state), + ), + } + }; + guest_wakeups.wake(); + host_wakeups.wake(); + guest.publish(); + host.publish(); + Ok(Ok(())) +} + +/// Reports the outcome of the socket `handle`'s vsock connect as a blocking +/// connect waiting for it sees it, as in Linux: +/// [`SocketConnectionStatus::Connecting`] while it is in progress, then +/// [`SocketConnectionStatus::Connected`], or its error, which stays pending +/// unless the connect timed out (Linux's waiter times out itself, and leaves +/// no error). An unconnected socket with no pending error is +/// [`SocketConnectionStatus::Unconnected`], and a listening one fails with +/// [`UnixSocketError::InvalidArgument`]. +pub fn check_connect( + process: &BrokerProcess, + handle: ObjectHandle, +) -> Result> { + let lease = Lease::new(process.authorized_object(handle, ObjectRights::WRITE)?)?; + let socket = &lease.socket; + let (result, taken) = socket.state.lock().connect_outcome(); + if taken { + socket.publish(); + } + Ok(result) +} + +/// Cancels the vsock connect in progress on the socket `handle`, as Linux +/// does when a blocking connect is interrupted: the socket is unconnected +/// again, still bound, with no pending error, and the connector's socket for +/// it hangs up, as if it had closed. +/// +/// With no connect in progress, it reports the outcome as [`check_connect`] +/// does: it fails with [`UnixSocketError::AlreadyConnected`] if the connect +/// completed, or with its error if it failed, and succeeds if the socket is +/// unconnected with no pending error. +pub fn cancel_vsock_connect( + process: &BrokerProcess, + handle: ObjectHandle, +) -> Result> { + let lease = Lease::new(process.authorized_object(handle, ObjectRights::WRITE)?)?; + let socket = &lease.socket; + let (hung_up, wakeups) = loop { + let peer = { + let mut state = socket.state.lock(); + let Connection::Connecting { peer, .. } = &state.connection else { + let (result, taken) = state.connect_outcome(); + drop(state); + if taken { + socket.publish(); + } + return Ok(match result { + Ok(SocketConnectionStatus::Connected) => Err(UnixSocketError::AlreadyConnected), + result => result.map(drop), + }); + }; + let Some(peer) = peer.upgrade() else { + break (None, socket.end_connect(&mut state, None)); + }; + peer + }; + let (mut state, mut peer_state) = lock_both(socket, &peer); + // Both ends are detached under both locks: a hang-up the connector's + // socket sends later finds no peer, and one it already sent finds no + // connect from it. + if matches!(&state.connection, Connection::Connecting { peer: current, .. } + if refers_to(current, &peer)) + { + let wakeups = socket.end_connect(&mut state, None); + let peer_wakeups = match &peer_state.connection { + Connection::Connected { peer: back, .. } if refers_to(back, socket) => { + peer.end_connection(&mut peer_state) + } + _ => Wakeups::default(), + }; + drop((state, peer_state)); + break (Some((peer, peer_wakeups)), wakeups); + } + }; + wakeups.wake(); + if let Some((peer, wakeups)) = hung_up { + wakeups.wake(); + peer.publish(); + } + socket.publish(); + Ok(Ok(())) +} + +/// Takes the pending error of a Unix socket: why its last connect failed. +pub fn take_error( + process: &BrokerProcess, + handle: ObjectHandle, +) -> Result> { + let lease = Lease::new( + process + .authorized_object_with_any_rights(handle, ObjectRights::WAIT | ObjectRights::WRITE)?, + )?; + let error = lease.socket.state.lock().error.take(); + if error.is_some() { + lease.socket.publish(); + } + Ok(error) } /// Accepts the oldest queued connection of a listening stream or @@ -333,7 +706,8 @@ pub fn accept( /// A stream socket sends as many bytes as its peer has room for. A datagram /// socket sends `data` as one datagram. A sequenced-packet socket sends /// `data` as one record to its peer, ignoring `address` like Linux. Fails -/// with [`BrokerError::WouldBlock`] while the receiving socket is full. +/// with [`BrokerError::WouldBlock`] while the receiving socket is full. A +/// datagram to a vsock address is [`UnixSocketError::Unreachable`]. pub fn send( process: &BrokerProcess, handle: ObjectHandle, @@ -410,7 +784,9 @@ pub fn receive( /// Shuts down one or both directions of a Unix socket. /// /// Shutting down a direction of a connected stream or sequenced-packet -/// socket also shuts down the opposite direction of its peer. +/// socket also shuts down the opposite direction of its peer. As in Linux, a +/// vsock socket that is not connected fails with +/// [`UnixSocketError::NotConnected`] and is not changed. pub fn shutdown( process: &BrokerProcess, handle: ObjectHandle, @@ -424,6 +800,9 @@ pub fn shutdown( }; let lease = Lease::new(process.authorized_object(handle, ObjectRights::WRITE)?)?; let socket = &lease.socket; + if let Err(error) = socket.shut_down_locally(read) { + return Ok(Err(error)); + } let (peer, wakeups) = socket.shut_down(read, write); wakeups.wake(); socket.publish(); @@ -436,7 +815,7 @@ pub fn shutdown( } /// Returns the name of a Unix socket, or of its connected peer if `peer` -/// is set. +/// is set. A vsock socket that shut down has no peer name. pub fn name( process: &BrokerProcess, handle: ObjectHandle, @@ -451,8 +830,13 @@ pub fn name( return Ok(Ok(state.name.clone())); } match &state.connection { - Connection::Connected { peer_name, .. } => Ok(Ok(peer_name.clone())), - Connection::None | Connection::Listening { .. } => Ok(Err(UnixSocketError::NotConnected)), + Connection::Connected { peer_name, .. } if state.local_shutdown.is_none() => { + Ok(Ok(peer_name.clone())) + } + Connection::Connected { .. } + | Connection::Connecting { .. } + | Connection::None + | Connection::Listening { .. } => Ok(Err(UnixSocketError::NotConnected)), } } @@ -500,15 +884,20 @@ fn creation_rights( /// /// A path resolves to its node, which requires write permission like Linux. /// The node is returned open, so callers keep its key from naming another -/// node while they use it. +/// node while they use it. A vsock address is unreachable: guests do not +/// connect or send to vsock sockets. fn lookup( process: &BrokerProcess, address: &UnixSocketAddress, user: FileUser, socket_type: SocketType, ) -> Result)>> { - let UnixSocketAddress::Path { path, .. } = address else { - return Ok(Ok((NameKey::of(address, socket_type), None))); + let path = match address { + UnixSocketAddress::Path { path, .. } => path, + UnixSocketAddress::Abstract(bytes) => { + return Ok(Ok((NameKey::Abstract(socket_type, bytes.clone()), None))); + } + UnixSocketAddress::Vsock { .. } => return Ok(Err(UnixSocketError::Unreachable)), }; // A path-only open does not copy the node up an overlay, so write // permission is checked against its status instead. @@ -556,8 +945,14 @@ pub(crate) struct UnixSocketObject { } impl UnixSocketObject { - /// Creates a new unnamed, unconnected `socket_type` socket. - fn new(process: &BrokerProcess, socket_type: SocketType, status: StatusFlags) -> Result { + /// Creates a new unnamed, unconnected `socket_type` socket, a vsock one + /// if `vsock` is set. + fn new( + process: &BrokerProcess, + socket_type: SocketType, + status: StatusFlags, + vsock: bool, + ) -> Result { if !matches!( socket_type, SocketType::Stream | SocketType::Datagram | SocketType::SeqPacket @@ -570,7 +965,8 @@ impl UnixSocketObject { Arc::clone(&process.unix_socket_bytes), UnixSocketName::Unnamed, Connection::None, - ); + ) + .with_vsock(vsock); Ok(Self::from_socket(Arc::new(socket), status)) } @@ -635,8 +1031,7 @@ impl Lease { /// The Unix sockets of a broker: the names they hold, and the charge for the /// data queued in them. pub(crate) struct UnixSockets { - /// Socket holding each bound name. - names: RwLock>>, + registry: RwLock, /// Bytes charged for data queued in every socket. Processes' charges /// change only under this lock too, so a charge commits both or neither. queued: Mutex, @@ -647,7 +1042,10 @@ pub(crate) struct UnixSockets { impl UnixSockets { pub(crate) fn new(limits: &BrokerCoreLimits) -> Self { Self { - names: RwLock::new(HashMap::new()), + registry: RwLock::new(Registry { + names: HashMap::new(), + next_ephemeral_vsock_port: None, + }), queued: Mutex::new(0), max_queued: limits.max_total_unix_socket_bytes, max_queued_per_process: limits.max_unix_socket_bytes_per_process, @@ -658,8 +1056,9 @@ impl UnixSockets { /// `socket_type`. fn find(&self, key: &NameKey, socket_type: SocketType) -> UnixSocketResult> { let socket = self - .names + .registry .read() + .names .get(key) .and_then(Weak::upgrade) .ok_or(UnixSocketError::ConnectionRefused)?; @@ -669,9 +1068,31 @@ impl UnixSockets { Ok(socket) } + /// Starts the ephemeral vsock ports at a random one, like Linux, so they + /// are unlikely to take the well-known ports a program binds later. The + /// seed only spreads the ports and is no secret, so without randomness + /// they start at the first. + /// + /// The caller holds no lock: the random provider may block. + fn seed_ephemeral_vsock_ports(&self, process: &BrokerProcess) { + if self.registry.read().next_ephemeral_vsock_port.is_some() { + return; + } + let mut seed = [0; 4]; + let offset = match crate::random::fill(process, &mut seed) { + Ok(()) => u32::from_ne_bytes(seed) % EPHEMERAL_VSOCK_PORTS, + Err(_) => 0, + }; + // Another bind may have seeded it, and taken ports, meanwhile. + self.registry + .write() + .next_ephemeral_vsock_port + .get_or_insert(offset); + } + /// Releases the name `key` if `socket` still holds it. fn unregister(&self, key: &NameKey, socket: &Socket) { - let mut names = self.names.write(); + let names = &mut self.registry.write().names; if names .get(key) .is_some_and(|holder| core::ptr::eq(holder.as_ptr(), socket)) @@ -712,6 +1133,26 @@ impl UnixSockets { } } +/// The names a broker's Unix sockets hold. +struct Registry { + /// Socket holding each bound name. + names: HashMap>, + /// Offset from [`FIRST_EPHEMERAL_VSOCK_PORT`] of the next ephemeral vsock + /// port to try, below [`EPHEMERAL_VSOCK_PORTS`], once seeded. + next_ephemeral_vsock_port: Option, +} + +impl Registry { + /// Takes the next ephemeral vsock port to try. The caller seeded the + /// ports ([`UnixSockets::seed_ephemeral_vsock_ports`]); unseeded, they + /// start at the first. + fn take_ephemeral_vsock_port(&mut self) -> u32 { + let offset = self.next_ephemeral_vsock_port.unwrap_or(0) % EPHEMERAL_VSOCK_PORTS; + self.next_ephemeral_vsock_port = Some((offset + 1) % EPHEMERAL_VSOCK_PORTS); + FIRST_EPHEMERAL_VSOCK_PORT + offset + } +} + /// Identity of a bound name. #[derive(Clone, Debug, PartialEq, Eq, Hash)] enum NameKey { @@ -720,16 +1161,10 @@ enum NameKey { /// An abstract name, which sockets of each type hold separately like /// Linux. Abstract(SocketType, Vec), -} - -impl NameKey { - /// Returns the key of an abstract address held by a `socket_type` socket. - fn of(address: &UnixSocketAddress, socket_type: SocketType) -> Self { - match address { - UnixSocketAddress::Abstract(bytes) => Self::Abstract(socket_type, bytes.clone()), - UnixSocketAddress::Path { .. } => unreachable!("path names are keyed by their node"), - } - } + /// A guest's vsock port, which only stream sockets hold. + Vsock(u32), + /// The host connector, which no address names. + VsockHost, } /// One Unix socket. @@ -739,6 +1174,9 @@ impl NameKey { /// to it may drop anywhere. struct Socket { kind: SocketType, + /// Whether the socket was created as a vsock socket ([`create_vsock`]); + /// one named by a vsock address is one too (see [`Socket::is_vsock`]). + vsock: bool, sockets: Arc, /// Quota of the process that created the socket, charged for the data /// queued for it and for its backlog. @@ -764,13 +1202,72 @@ struct State { /// Whether the socket stopped sending. Its incoming channel is shut once /// it stops receiving. write_shut: bool, + /// How a vsock socket's own shutdown, not its peer's, stopped it. + local_shutdown: Option, + /// Why the socket's last connect failed, if no call reported it yet; + /// only while it is unconnected. + error: Option, /// Whether the socket was torn down. closed: bool, connection: Connection, } +impl State { + /// Checks that a stream or sequenced-packet socket in this state may + /// connect to a vsock address; else fails as Linux does, taking the + /// pending error, if any. + fn check_vsock_connect(&mut self) -> UnixSocketResult<()> { + match self.connection { + Connection::Connected { .. } if self.local_shutdown.is_some() => { + Err(UnixSocketError::InvalidArgument) + } + Connection::Connected { .. } => Err(UnixSocketError::AlreadyConnected), + Connection::Connecting { .. } => Err(UnixSocketError::AlreadyInProgress), + Connection::Listening { .. } => Err(UnixSocketError::InvalidArgument), + Connection::None => self.error.take().map_or(Ok(()), Err), + } + } + + /// Returns the outcome of a vsock connect as [`check_connect`] reports + /// it, and whether it took the pending error. + fn connect_outcome(&mut self) -> (UnixSocketResult, bool) { + let result = match self.connection { + Connection::Connecting { .. } => Ok(SocketConnectionStatus::Connecting), + Connection::Connected { .. } => Ok(SocketConnectionStatus::Connected), + Connection::Listening { .. } => Err(UnixSocketError::InvalidArgument), + Connection::None => match self.error { + Some(UnixSocketError::TimedOut) => { + self.error = None; + return (Err(UnixSocketError::TimedOut), true); + } + Some(error) => Err(error), + None => Ok(SocketConnectionStatus::Unconnected), + }, + }; + (result, false) + } +} + +/// What a vsock socket's own shutdown stopped. +#[derive(Clone, Copy, PartialEq, Eq)] +enum LocalShutdown { + /// Sending only. + Write, + /// Receiving, and maybe sending. + Read, +} + enum Connection { None, + /// Connecting to the host's vsock port through the host connector (see + /// the module docs), as a connected socket would be but for its readiness + /// and operations. + Connecting { + /// The connector's socket for the connection, which completes it. + peer: Weak, + /// The host's name, which the connection will have. + peer_name: UnixSocketName, + }, Listening { /// Connected sockets not yet accepted. backlog: VecDeque>, @@ -827,8 +1324,8 @@ impl Datagram { /// /// A weak reference keeps its allocation, so no other socket can take its /// address while it exists. -fn refers_to(weak: &Weak, socket: &Arc) -> bool { - core::ptr::eq(weak.as_ptr(), Arc::as_ptr(socket)) +fn refers_to(weak: &Weak, socket: &Socket) -> bool { + core::ptr::eq(weak.as_ptr(), socket) } /// Locks the states of two different sockets in address order, so connects @@ -853,6 +1350,11 @@ impl Waiter for Socket { } impl Socket { + /// Whether the socket, whose state is `state`, has vsock semantics. + fn is_vsock(&self, state: &State) -> bool { + self.vsock || matches!(state.name, UnixSocketName::Vsock { .. }) + } + fn new( kind: SocketType, sockets: Arc, @@ -862,6 +1364,7 @@ impl Socket { ) -> Self { Self { kind, + vsock: false, sockets, quota, state: Mutex::new(State { @@ -871,6 +1374,8 @@ impl Socket { marker: None, options: SocketOptions::default(), write_shut: false, + local_shutdown: None, + error: None, closed: false, connection, }), @@ -879,6 +1384,11 @@ impl Socket { } } + /// Makes the socket a vsock socket if `vsock` is set. + fn with_vsock(self, vsock: bool) -> Self { + Self { vsock, ..self } + } + /// Publishes the socket's readiness after a change that may affect it. fn publish(self: &Arc) { let watchers = self.watchers.lock(); @@ -898,16 +1408,29 @@ impl Socket { /// reports itself writable, so a send retries and reports the failure /// rather than waiting for a wakeup that never comes. fn readiness(self: &Arc) -> ReadinessFlags { - let (write_shut, connection) = { + let (write_shut, error, vsock, connecting, connection) = { let state = self.state.lock(); let connection = match &state.connection { Connection::None => None, Connection::Listening { backlog, .. } => Some(Err(!backlog.is_empty())), + // Neither readable nor writable until it completes. + Connection::Connecting { .. } => Some(Err(false)), Connection::Connected { peer, .. } => Some(Ok(peer.clone())), }; - (state.write_shut, connection) + let vsock = self.is_vsock(&state); + let connecting = matches!(state.connection, Connection::Connecting { .. }); + ( + state.write_shut, + state.error.is_some(), + vsock, + connecting, + connection, + ) }; let mut readiness = ReadinessFlags::default(); + if error { + readiness = readiness | ReadinessFlags::ERROR; + } { let channel = self.incoming.lock(); if channel.is_shut() { @@ -916,13 +1439,16 @@ impl Socket { readiness = readiness | ReadinessFlags::CLOSED; } } - if !channel.queue.is_empty() { + // A connecting socket receives nothing until it connects. + if !connecting && !channel.queue.is_empty() { readiness = readiness | ReadinessFlags::READ; } } let peer = match connection { None => { - if self.is_connection_oriented() { + // An unconnected vsock socket is not hung up (see the module + // docs). + if self.is_connection_oriented() && !vsock { readiness = readiness | ReadinessFlags::CLOSED; } None @@ -951,15 +1477,50 @@ impl Socket { key: NameKey, name: UnixSocketName, ) -> Result> { - let mut names = self.sockets.names.write(); - if names.contains_key(&key) { + let mut registry = self.sockets.registry.write(); + if registry.names.contains_key(&key) { return Ok(Err(UnixSocketError::AddressInUse)); } + self.hold_name(&mut registry.names, state, key, name) + .map(Ok) + } + + /// Records that the socket holds an unused ephemeral vsock port. + fn register_ephemeral_vsock_port( + self: &Arc, + state: &mut State, + ) -> Result> { + let mut registry = self.sockets.registry.write(); + for _ in 0..MAX_EPHEMERAL_VSOCK_PORT_ATTEMPTS { + let port = registry.take_ephemeral_vsock_port(); + let key = NameKey::Vsock(port); + if !registry.names.contains_key(&key) { + let name = UnixSocketName::Vsock { + cid: VMADDR_CID_ANY, + port, + }; + return self + .hold_name(&mut registry.names, state, key, name) + .map(Ok); + } + } + Ok(Err(UnixSocketError::AddressNotAvailable)) + } + + /// Records that the socket holds the unused name `name` identified by + /// `key` in `names`, the locked registry. + fn hold_name( + self: &Arc, + names: &mut HashMap>, + state: &mut State, + key: NameKey, + name: UnixSocketName, + ) -> Result<()> { names.try_reserve(1).map_err(|_| BrokerError::OutOfMemory)?; names.insert(key.clone(), Arc::downgrade(self)); state.name = name; state.name_key = Some(key); - Ok(Ok(())) + Ok(()) } /// Queues a connection to `listener`, which accepts it as a new socket of @@ -973,6 +1534,7 @@ impl Socket { // A socket never listens while connecting. return Ok(Err(match self.state.lock().connection { Connection::None => UnixSocketError::ConnectionRefused, + Connection::Connecting { .. } => UnixSocketError::AlreadyInProgress, Connection::Connected { .. } => UnixSocketError::AlreadyConnected, Connection::Listening { .. } => UnixSocketError::InvalidArgument, })); @@ -985,6 +1547,9 @@ impl Socket { } match state.connection { Connection::None => {} + Connection::Connecting { .. } => { + return Ok(Err(UnixSocketError::AlreadyInProgress)); + } Connection::Connected { .. } => return Ok(Err(UnixSocketError::AlreadyConnected)), Connection::Listening { .. } => return Ok(Err(UnixSocketError::InvalidArgument)), } @@ -1002,31 +1567,111 @@ impl Socket { return Err(BrokerError::would_block(nonblocking)); } } - backlog - .try_reserve(1) - .map_err(|_| BrokerError::OutOfMemory)?; - self.sockets.charge(&listener.quota, CONNECTION_OVERHEAD)?; - let server = Arc::new(Socket::new( - self.kind, - Arc::clone(&listener.sockets), - Arc::clone(&listener.quota), - listener_name.clone(), - Connection::Connected { - peer: Arc::downgrade(self), - peer_name: state.name.clone(), - }, - )); + let peer = self.queue_on(&state, listener, backlog, listener_name.clone())?; state.connection = Connection::Connected { - peer: Arc::downgrade(&server), + peer, peer_name: listener_name, }; - backlog.push_back(server); } listener.publish(); self.publish(); Ok(Ok(())) } + /// Starts this stream or sequenced-packet socket's connect to vsock + /// address `cid`:`port` (see [`connect`]). + fn connect_to_vsock_host( + self: &Arc, + process: &BrokerProcess, + cid: u32, + port: u32, + ) -> Result> { + let unnamed = { + let mut state = self.state.lock(); + if let Err(error) = state.check_vsock_connect() { + return Ok(Err(error)); + } + state.name == UnixSocketName::Unnamed + }; + if self.kind != SocketType::Stream || cid != VMADDR_CID_HOST { + return Ok(Err(UnixSocketError::Unreachable)); + } + let connector = match self.sockets.find(&NameKey::VsockHost, SocketType::Stream) { + // The connector listens, so this socket is not it. + Ok(connector) if !Arc::ptr_eq(self, &connector) => connector, + _ => return Ok(Err(UnixSocketError::Unreachable)), + }; + if unnamed { + self.sockets.seed_ephemeral_vsock_ports(process); + } + { + let (mut state, mut connecting) = lock_both(self, &connector); + // Another call may have connected or named it meanwhile. + if let Err(error) = state.check_vsock_connect() { + return Ok(Err(error)); + } + match state.name { + UnixSocketName::Unnamed if !state.binding => { + if let Err(error) = self.register_ephemeral_vsock_port(&mut state)? { + return Ok(Err(error)); + } + } + UnixSocketName::Vsock { + cid: VMADDR_CID_ANY, + .. + } => {} + // Not a vsock socket. + _ => return Ok(Err(UnixSocketError::Unreachable)), + } + if connecting.closed { + return Ok(Err(UnixSocketError::Unreachable)); + } + let Connection::Listening { backlog, limit } = &mut connecting.connection else { + return Ok(Err(UnixSocketError::Unreachable)); + }; + if backlog.len() > *limit { + return Err(BrokerError::ResourceExhausted); + } + let peer_name = UnixSocketName::Vsock { + cid: VMADDR_CID_HOST, + port, + }; + let peer = self.queue_on(&state, &connector, backlog, peer_name.clone())?; + state.connection = Connection::Connecting { peer, peer_name }; + } + connector.publish(); + self.publish(); + Ok(Ok(SocketConnectionStatus::Connecting)) + } + + /// Queues on `backlog`, `listener`'s, a new socket named `name` and + /// connected to this one, whose state is `state`; returns it. + fn queue_on( + self: &Arc, + state: &State, + listener: &Socket, + backlog: &mut VecDeque>, + name: UnixSocketName, + ) -> Result> { + backlog + .try_reserve(1) + .map_err(|_| BrokerError::OutOfMemory)?; + self.sockets.charge(&listener.quota, CONNECTION_OVERHEAD)?; + let server = Arc::new(Socket::new( + self.kind, + Arc::clone(&listener.sockets), + Arc::clone(&listener.quota), + name, + Connection::Connected { + peer: Arc::downgrade(self), + peer_name: state.name.clone(), + }, + )); + let peer = Arc::downgrade(&server); + backlog.push_back(server); + Ok(peer) + } + fn connect_datagram(self: &Arc, target: &Arc) -> UnixSocketResult<()> { // Both states stay locked from the check to the connection, like // Linux, so neither socket connects elsewhere meanwhile. @@ -1149,7 +1794,9 @@ impl Socket { let state = self.state.lock(); let peer = match &state.connection { Connection::Connected { peer, .. } => Some(peer.clone()), - Connection::None | Connection::Listening { .. } => None, + Connection::None | Connection::Connecting { .. } | Connection::Listening { .. } => { + None + } }; (peer, state.write_shut, state.name.clone()) }; @@ -1295,13 +1942,19 @@ impl Socket { source: bool, nonblocking: bool, ) -> Result> { - let source = match &self.state.lock().connection { - Connection::Connected { peer_name, .. } => source.then(|| peer_name.clone()), - Connection::None | Connection::Listening { .. } => { - return Ok(Err(UnixSocketError::InvalidArgument)); - } + let (source, read_shut) = { + let state = self.state.lock(); + let source = match &state.connection { + Connection::Connected { peer_name, .. } => source.then(|| peer_name.clone()), + Connection::Connecting { .. } => return Ok(Err(UnixSocketError::NotConnected)), + Connection::None | Connection::Listening { .. } => { + return Ok(Err(UnixSocketError::InvalidArgument)); + } + }; + (source, state.local_shutdown == Some(LocalShutdown::Read)) }; - if output.is_empty() { + // Its incoming channel is shut too, so it reports itself readable. + if output.is_empty() || read_shut { return Ok(Ok(Received { source, ..Received::default() @@ -1410,6 +2063,27 @@ impl Socket { self.receive_datagram(output, peek, source, nonblocking) } + /// Records that a vsock socket's own shutdown stops it receiving if `read` + /// is set, or else sending. Fails, recording nothing, if it is a vsock + /// socket that is not connected. + fn shut_down_locally(&self, read: bool) -> UnixSocketResult<()> { + let mut state = self.state.lock(); + if !self.is_vsock(&state) { + return Ok(()); + } + // A connected stream socket stays connected while referenced, so the + // shutdown that follows the check finds it so. + if !matches!(state.connection, Connection::Connected { .. }) { + return Err(UnixSocketError::NotConnected); + } + if read { + state.local_shutdown = Some(LocalShutdown::Read); + } else { + state.local_shutdown.get_or_insert(LocalShutdown::Write); + } + Ok(()) + } + /// Stops the socket from receiving if `read` is set, and from sending if /// `write` is set, returning its connected peer if the peer shares the /// shutdown. @@ -1430,22 +2104,53 @@ impl Socket { } /// Ends this connection-oriented socket's connection after its peer - /// closed. - fn hang_up(self: &Arc) { + /// `from` closed; a connect still in progress fails instead. + /// + /// Does nothing unless `from` is still its peer: a hang-up sent after the + /// connection was detached (see [`cancel_vsock_connect`]) must leave the + /// unconnected socket, or its next connection, alone. + fn hang_up(self: &Arc, from: &Socket) { let wakeups = { let mut state = self.state.lock(); - state.write_shut = true; - if let Connection::Connected { peer, .. } = &mut state.connection { - *peer = Weak::new(); + match &state.connection { + Connection::Connecting { peer, .. } if refers_to(peer, from) => { + self.end_connect(&mut state, Some(UnixSocketError::ConnectionReset)) + } + Connection::Connected { peer, .. } if refers_to(peer, from) => { + self.end_connection(&mut state) + } + _ => return, } - let mut channel = self.incoming.lock(); - channel.shut(); - channel.wake() }; wakeups.wake(); self.publish(); } + /// Ends this socket's connection, whose state is `state`, after its peer + /// left: it no longer sends, and receives the end of data once its queue + /// drains. + fn end_connection(&self, state: &mut State) -> Wakeups { + state.write_shut = true; + if let Connection::Connected { peer, .. } = &mut state.connection { + *peer = Weak::new(); + } + let mut channel = self.incoming.lock(); + channel.shut(); + channel.wake() + } + + /// Leaves this connecting socket, whose state is `state`, unconnected, + /// with the pending error `error`, if any, and discards the data its peer + /// queued for it meanwhile, which no later connection may receive. + fn end_connect(&self, state: &mut State, error: Option) -> Wakeups { + state.connection = Connection::None; + state.error = error; + let mut channel = self.incoming.lock(); + self.purge(&mut channel); + // Senders waiting for room retry. + channel.wake() + } + /// Discards the data queued for the socket. fn purge(&self, channel: &mut ChannelState) { channel.queue = Queue::default(); @@ -1487,12 +2192,14 @@ impl Socket { socket.close(); } } - Connection::Connected { peer, .. } if self.is_connection_oriented() => { + Connection::Connected { peer, .. } | Connection::Connecting { peer, .. } + if self.is_connection_oriented() => + { if let Some(peer) = peer.upgrade() { - peer.hang_up(); + peer.hang_up(self); } } - Connection::None | Connection::Connected { .. } => {} + Connection::None | Connection::Connecting { .. } | Connection::Connected { .. } => {} } wakeups.wake(); } diff --git a/litebox_broker_core/src/unix_socket/tests.rs b/litebox_broker_core/src/unix_socket/tests.rs index 05ee54c1cf..8f6bee4d12 100644 --- a/litebox_broker_core/src/unix_socket/tests.rs +++ b/litebox_broker_core/src/unix_socket/tests.rs @@ -63,12 +63,29 @@ impl Fixture { } fn build(limits: BrokerCoreLimits, rights: ObjectRights) -> Self { + Self::build_with_random(limits, rights, Arc::new(crate::random::TestRandomProvider)) + } + + fn without_randomness() -> Self { + Self::build_with_random( + BrokerCoreLimits::DEFAULT, + ObjectRights::all(), + Arc::new(crate::test_support::FailingRandomProvider), + ) + } + + fn build_with_random( + limits: BrokerCoreLimits, + rights: ObjectRights, + random_provider: Arc, + ) -> Self { let fs = Resolver::::new(InMem::::new( InodeAllocator::standalone(), )); let broker = TestBrokerCoreBuilder::new(PolicyEngine::with_unauthenticated_rights(rights)) .with_limits(limits) .with_file_service(Arc::new(fs)) + .with_random_provider(random_provider) .build() .unwrap(); let process = broker @@ -123,12 +140,14 @@ impl Fixture { listener } + /// Connects `handle` to `address`, which is not the host's. fn connect( &self, handle: ObjectHandle, address: &UnixSocketAddress, ) -> Result> { - connect(&self.process, handle, address, USER) + let result = connect(&self.process, handle, address, USER)?; + Ok(result.map(|status| assert_eq!(status, SocketConnectionStatus::Connected))) } fn accept(&self, listener: ObjectHandle) -> Result> { @@ -216,6 +235,31 @@ fn abstract_name(name: &[u8]) -> UnixSocketAddress { UnixSocketAddress::Abstract(name.to_vec()) } +fn vsock(port: u32) -> UnixSocketAddress { + UnixSocketAddress::Vsock { + cid: VMADDR_CID_ANY, + port, + } +} + +fn guest_vsock_name(port: u32) -> UnixSocketName { + UnixSocketName::Vsock { + cid: VMADDR_CID_ANY, + port, + } +} + +impl Fixture { + fn connect_from_host( + &self, + handle: ObjectHandle, + guest_port: u32, + host_port: u32, + ) -> Result> { + connect_from_vsock_host(&self.process, handle, guest_port, host_port) + } +} + #[test] fn stream_pairs_carry_bytes_in_both_directions() { let fixture = Fixture::new(); @@ -385,7 +429,7 @@ fn path_connections_require_write_permission() { .unwrap(); assert_eq!( connect(&fixture.process, client, &address, other).unwrap(), - Ok(()) + Ok(SocketConnectionStatus::Connected) ); assert!(fixture.accept(listener).unwrap().is_ok()); @@ -500,12 +544,23 @@ fn closing_a_listener_resets_unaccepted_connections() { .connection { Connection::Connected { peer, .. } => peer.clone(), - Connection::None | Connection::Listening { .. } => panic!("the client is connected"), + Connection::None | Connection::Connecting { .. } | Connection::Listening { .. } => { + panic!("the client is connected") + } }; fixture.close(listener); assert_eq!(fixture.queued(), (0, 0)); assert!(server.upgrade().is_none()); - assert!(fixture.process.core.unix_sockets.names.read().is_empty()); + assert!( + fixture + .process + .core + .unix_sockets + .registry + .read() + .names + .is_empty() + ); assert!( fixture .readiness(client) @@ -1181,3 +1236,1161 @@ fn opposite_connects_do_not_deadlock() { } assert_eq!(fixture.queued(), (0, 0)); } + +#[test] +fn vsock_ports_are_apart_from_unix_names() { + let fixture = Fixture::new(); + let unix = fixture.listener(&abstract_name(b"vsock:1024"), 1); + let native = fixture.listener(&vsock(1024), 1); + assert_eq!( + name(&fixture.process, native, false).unwrap(), + Ok(guest_vsock_name(1024)) + ); + + // A Unix connection reaches only the Unix listener, and the host's only + // the vsock one. + let client = fixture.create(SocketType::Stream); + assert_eq!( + fixture + .connect(client, &abstract_name(b"vsock:1024")) + .unwrap(), + Ok(()) + ); + let host = fixture.create(SocketType::Stream); + assert_eq!(fixture.connect_from_host(host, 1024, 5).unwrap(), Ok(())); + assert!(fixture.accept(unix).unwrap().is_ok()); + assert!(fixture.accept(native).unwrap().is_ok()); + assert_eq!(fixture.accept(unix), Err(BrokerError::WouldBlock)); + assert_eq!(fixture.accept(native), Err(BrokerError::WouldBlock)); + + // Nor does the host reach a Unix listener. + fixture.close(native); + let host = fixture.create(SocketType::Stream); + assert_eq!( + fixture.connect_from_host(host, 1024, 5).unwrap(), + Err(UnixSocketError::ConnectionRefused) + ); +} + +#[test] +fn only_stream_sockets_bind_vsock_ports_at_any_cid() { + let fixture = Fixture::new(); + for socket_type in [SocketType::Datagram, SocketType::SeqPacket] { + let socket = fixture.create(socket_type); + assert_eq!( + fixture.bind(socket, &vsock(1024)), + Err(UnixSocketError::InvalidArgument) + ); + } + let socket = fixture.create(SocketType::Stream); + for cid in [0, 1, VMADDR_CID_HOST, 3] { + assert_eq!( + fixture.bind(socket, &UnixSocketAddress::Vsock { cid, port: 1024 }), + Err(UnixSocketError::AddressNotAvailable) + ); + } + assert_eq!(fixture.bind(socket, &vsock(1024)), Ok(())); + assert_eq!( + fixture.bind(socket, &vsock(1025)), + Err(UnixSocketError::InvalidArgument) + ); + let other = fixture.create(SocketType::Stream); + assert_eq!( + fixture.bind(other, &vsock(1024)), + Err(UnixSocketError::AddressInUse) + ); + fixture.close(socket); + assert_eq!(fixture.bind(other, &vsock(1024)), Ok(())); +} + +#[test] +fn vsock_port_any_binds_an_unused_ephemeral_port() { + let fixture = Fixture::new(); + let bound_port = |handle| match name(&fixture.process, handle, false).unwrap() { + Ok(UnixSocketName::Vsock { + cid: VMADDR_CID_ANY, + port, + }) => port, + other => panic!("not a guest vsock name: {other:?}"), + }; + let first = fixture.create(SocketType::Stream); + assert_eq!(fixture.bind(first, &vsock(VMADDR_PORT_ANY)), Ok(())); + let first_port = bound_port(first); + assert!((FIRST_EPHEMERAL_VSOCK_PORT..VMADDR_PORT_ANY).contains(&first_port)); + + // The next is taken, so the one after it is bound. + let taken = fixture.create(SocketType::Stream); + let next = if first_port == VMADDR_PORT_ANY - 1 { + FIRST_EPHEMERAL_VSOCK_PORT + } else { + first_port + 1 + }; + fixture.bind(taken, &vsock(next)).unwrap(); + let second = fixture.create(SocketType::Stream); + assert_eq!(fixture.bind(second, &vsock(VMADDR_PORT_ANY)), Ok(())); + assert_ne!(bound_port(second), first_port); + assert_ne!(bound_port(second), next); + assert!(bound_port(second) >= FIRST_EPHEMERAL_VSOCK_PORT); + let listener = fixture.create(SocketType::Stream); + fixture.bind(listener, &vsock(VMADDR_PORT_ANY)).unwrap(); + listen(&fixture.process, listener, 1).unwrap().unwrap(); + let host = fixture.create(SocketType::Stream); + assert_eq!( + fixture + .connect_from_host(host, bound_port(listener), 5) + .unwrap(), + Ok(()) + ); +} + +#[test] +fn ephemeral_vsock_ports_wrap_and_give_up_when_taken() { + let fixture = Fixture::new(); + let sockets = &fixture.process.core.unix_sockets; + sockets.registry.write().next_ephemeral_vsock_port = Some(EPHEMERAL_VSOCK_PORTS - 1); + let last = fixture.create(SocketType::Stream); + fixture.bind(last, &vsock(VMADDR_PORT_ANY)).unwrap(); + assert_eq!( + name(&fixture.process, last, false).unwrap(), + Ok(guest_vsock_name(VMADDR_PORT_ANY - 1)) + ); + + // Every port the next bind tries is taken. + for offset in 0..MAX_EPHEMERAL_VSOCK_PORT_ATTEMPTS { + let socket = fixture.create(SocketType::Stream); + fixture + .bind(socket, &vsock(FIRST_EPHEMERAL_VSOCK_PORT + offset)) + .unwrap(); + } + let socket = fixture.create(SocketType::Stream); + assert_eq!( + fixture.bind(socket, &vsock(VMADDR_PORT_ANY)), + Err(UnixSocketError::AddressNotAvailable) + ); + assert_eq!( + name(&fixture.process, socket, false).unwrap(), + Ok(UnixSocketName::Unnamed) + ); + assert_eq!(fixture.bind(socket, &vsock(VMADDR_PORT_ANY)), Ok(())); +} + +#[test] +fn only_ephemeral_binds_seed_vsock_ports_and_need_no_randomness() { + let fixture = Fixture::without_randomness(); + let unseeded = || { + fixture + .process + .core + .unix_sockets + .registry + .read() + .next_ephemeral_vsock_port + .is_none() + }; + let datagram = fixture.create(SocketType::Datagram); + assert_eq!( + fixture.bind(datagram, &vsock(VMADDR_PORT_ANY)), + Err(UnixSocketError::InvalidArgument) + ); + let socket = fixture.create(SocketType::Stream); + assert_eq!( + fixture.bind( + socket, + &UnixSocketAddress::Vsock { + cid: 3, + port: VMADDR_PORT_ANY + } + ), + Err(UnixSocketError::AddressNotAvailable) + ); + fixture.bind(socket, &vsock(2048)).unwrap(); + assert_eq!( + fixture.bind(socket, &vsock(VMADDR_PORT_ANY)), + Err(UnixSocketError::InvalidArgument) + ); + assert!(unseeded()); + + // Without randomness, the ports start at the first. + let ephemeral = fixture.create(SocketType::Stream); + assert_eq!(fixture.bind(ephemeral, &vsock(VMADDR_PORT_ANY)), Ok(())); + assert_eq!( + name(&fixture.process, ephemeral, false).unwrap(), + Ok(guest_vsock_name(FIRST_EPHEMERAL_VSOCK_PORT)) + ); + assert!(!unseeded()); +} + +#[test] +fn vsock_addresses_are_unreachable_without_a_host_connector() { + let fixture = Fixture::new(); + fixture.listener(&vsock(1024), 1); + for address in [ + vsock(1024), + UnixSocketAddress::Vsock { + cid: VMADDR_CID_HOST, + port: 1024, + }, + ] { + for socket_type in [SocketType::Stream, SocketType::Datagram] { + let client = fixture.create(socket_type); + assert_eq!( + fixture.connect(client, &address).unwrap(), + Err(UnixSocketError::Unreachable) + ); + } + let sender = fixture.create(SocketType::Datagram); + assert_eq!( + fixture.send_to(sender, &address, b"x").unwrap(), + Err(UnixSocketError::Unreachable) + ); + } +} + +#[test] +fn host_connections_bear_the_hosts_name() { + let fixture = Fixture::new(); + let listener = fixture.listener(&vsock(1024), 8); + let host = fixture.create(SocketType::Stream); + assert_eq!( + fixture.connect_from_host(host, 1024, 49152).unwrap(), + Ok(()) + ); + let host_name = UnixSocketName::Vsock { + cid: VMADDR_CID_HOST, + port: 49152, + }; + assert_eq!( + name(&fixture.process, host, false).unwrap(), + Ok(host_name.clone()) + ); + assert_eq!( + name(&fixture.process, host, true).unwrap(), + Ok(guest_vsock_name(1024)) + ); + assert_eq!( + fixture.connect_from_host(host, 1024, 49152).unwrap(), + Err(UnixSocketError::InvalidArgument) + ); + + let server = fixture.accept(listener).unwrap().unwrap(); + assert_eq!( + name(&fixture.process, server, false).unwrap(), + Ok(guest_vsock_name(1024)) + ); + assert_eq!( + name(&fixture.process, server, true).unwrap(), + Ok(host_name.clone()) + ); + assert_eq!(fixture.send(host, b"ping").unwrap(), Ok(4)); + let received = fixture.receive(server, 8).unwrap().unwrap(); + assert_eq!(received.data, b"ping"); + assert_eq!(received.source, Some(host_name)); + assert_eq!(fixture.send(server, b"pong").unwrap(), Ok(4)); + assert_eq!(fixture.receive_data(host, 8), b"pong"); + + // Host ports hold no name: a guest neither binds them nor finds them. + let other = fixture.create(SocketType::Stream); + assert_eq!(fixture.bind(other, &vsock(49152)), Ok(())); +} + +#[test] +fn host_connections_are_refused_or_wait_like_others() { + let fixture = Fixture::new(); + let host = fixture.create(SocketType::Stream); + assert_eq!( + fixture.connect_from_host(host, 1024, 5).unwrap(), + Err(UnixSocketError::ConnectionRefused) + ); + + // A full backlog does not wait. + let listener = fixture.listener(&vsock(1024), 0); + let first = fixture.create(SocketType::Stream); + assert_eq!(fixture.connect_from_host(first, 1024, 6).unwrap(), Ok(())); + let second = fixture.create(SocketType::Stream); + assert_eq!( + fixture.connect_from_host(second, 1024, 7), + Err(BrokerError::NonBlockingWouldBlock) + ); + fixture.accept(listener).unwrap().unwrap(); + let third = fixture.create(SocketType::Stream); + assert_eq!(fixture.connect_from_host(third, 1024, 7).unwrap(), Ok(())); + + // Only unnamed, unconnected streams connect. + let named = fixture.create(SocketType::Stream); + fixture.bind(named, &abstract_name(b"named")).unwrap(); + let (connected, _) = fixture.pair(SocketType::Stream); + for socket in [named, connected] { + assert_eq!( + fixture.connect_from_host(socket, 1024, 8).unwrap(), + Err(UnixSocketError::InvalidArgument) + ); + } + let datagram = fixture.create(SocketType::Datagram); + assert_eq!( + fixture.connect_from_host(datagram, 1024, 8).unwrap(), + Err(UnixSocketError::WrongType) + ); +} + +impl Fixture { + /// Returns a host connection from host port 5 to a new listener on vsock + /// port `port`, and the guest's accepted socket. + fn host_connection(&self, port: u32) -> (ObjectHandle, ObjectHandle) { + let listener = self.listener(&vsock(port), 1); + let host = self.create(SocketType::Stream); + self.connect_from_host(host, port, 5).unwrap().unwrap(); + let guest = self.accept(listener).unwrap().unwrap(); + self.close(listener); + (host, guest) + } +} + +#[test] +fn vsock_sockets_shut_down_receiving_drop_queued_data() { + let fixture = Fixture::new(); + for mode in [ShutdownMode::Read, ShutdownMode::Both] { + let (host, guest) = fixture.host_connection(1024); + assert_eq!(fixture.send(host, b"queued").unwrap(), Ok(6)); + assert_eq!( + receive_message(&fixture.process, guest, 8, true, false, false) + .unwrap() + .unwrap() + .data, + b"queued" + ); + shutdown(&fixture.process, guest, mode).unwrap().unwrap(); + assert!(fixture.readiness(guest).contains(ReadinessFlags::READ)); + for peek in [true, false] { + assert_eq!( + receive_message(&fixture.process, guest, 8, peek, false, true), + Ok(Ok(Message::default())) + ); + } + assert_eq!( + name(&fixture.process, guest, true).unwrap(), + Err(UnixSocketError::NotConnected) + ); + if mode == ShutdownMode::Read { + assert_eq!(fixture.send(guest, b"reply").unwrap(), Ok(5)); + assert_eq!(fixture.receive_data(host, 8), b"reply"); + } else { + assert_eq!( + fixture.send(guest, b"reply").unwrap(), + Err(UnixSocketError::BrokenPipe) + ); + } + } +} + +#[test] +fn vsock_sockets_drain_data_their_peer_sent_before_shutting_down() { + let fixture = Fixture::new(); + let (host, guest) = fixture.host_connection(1024); + assert_eq!(fixture.send(host, b"tail").unwrap(), Ok(4)); + shutdown(&fixture.process, host, ShutdownMode::Write) + .unwrap() + .unwrap(); + assert_eq!(fixture.receive_data(guest, 8), b"tail"); + assert_eq!(fixture.receive_data(guest, 8), b""); + // Only the socket that shut down loses its peer's name. + assert_eq!( + name(&fixture.process, guest, true).unwrap(), + Ok(UnixSocketName::Vsock { + cid: VMADDR_CID_HOST, + port: 5 + }) + ); + assert_eq!( + name(&fixture.process, host, true).unwrap(), + Err(UnixSocketError::NotConnected) + ); +} + +#[test] +fn vsock_sockets_shut_down_sending_keep_receiving() { + let fixture = Fixture::new(); + let (host, guest) = fixture.host_connection(1024); + assert_eq!(fixture.send(host, b"data").unwrap(), Ok(4)); + shutdown(&fixture.process, guest, ShutdownMode::Write) + .unwrap() + .unwrap(); + assert_eq!( + name(&fixture.process, guest, true).unwrap(), + Err(UnixSocketError::NotConnected) + ); + assert_eq!(fixture.receive_data(guest, 8), b"data"); + // A later shutdown of receiving still drops what is queued. + assert_eq!(fixture.send(host, b"late").unwrap(), Ok(4)); + shutdown(&fixture.process, guest, ShutdownMode::Read) + .unwrap() + .unwrap(); + assert_eq!(fixture.receive_data(guest, 8), b""); +} + +#[test] +fn unix_sockets_shut_down_receiving_drain_queued_data() { + let fixture = Fixture::new(); + let (first, second) = fixture.pair(SocketType::Stream); + assert_eq!(fixture.send(second, b"queued").unwrap(), Ok(6)); + shutdown(&fixture.process, first, ShutdownMode::Both) + .unwrap() + .unwrap(); + assert_eq!(fixture.receive_data(first, 8), b"queued"); + assert_eq!( + name(&fixture.process, first, true).unwrap(), + Ok(UnixSocketName::Unnamed) + ); +} + +#[test] +fn unconnected_vsock_sockets_do_not_shut_down() { + let fixture = Fixture::new(); + // Nor does one created as a vsock socket, before it has a name. + let created = create_vsock( + &fixture.process, + FileOpenFlags::NONE, + &fixture.readiness_sink, + ) + .unwrap(); + let bound = fixture.create(SocketType::Stream); + fixture.bind(bound, &vsock(1025)).unwrap(); + let listener = fixture.listener(&vsock(1024), 1); + for socket in [created, bound, listener] { + for mode in [ShutdownMode::Read, ShutdownMode::Write, ShutdownMode::Both] { + assert_eq!( + shutdown(&fixture.process, socket, mode).unwrap(), + Err(UnixSocketError::NotConnected) + ); + } + } + assert_eq!(fixture.readiness(created), ReadinessFlags::WRITE); + assert_eq!(fixture.readiness(listener), ReadinessFlags::default()); + fixture.bind(created, &vsock(1026)).unwrap(); + for socket in [created, bound] { + listen(&fixture.process, socket, 1).unwrap().unwrap(); + } + for (socket, port) in [(listener, 1024), (bound, 1025), (created, 1026)] { + let host = fixture.create(SocketType::Stream); + assert_eq!(fixture.connect_from_host(host, port, 5).unwrap(), Ok(())); + let guest = fixture.accept(socket).unwrap().unwrap(); + assert_eq!(fixture.send(host, b"ping").unwrap(), Ok(4)); + assert_eq!(fixture.receive_data(guest, 8), b"ping"); + } +} + +#[test] +fn unconnected_unix_sockets_still_shut_down() { + let fixture = Fixture::new(); + let unbound = fixture.create(SocketType::Stream); + let bound = fixture.create(SocketType::Stream); + fixture.bind(bound, &abstract_name(b"bound")).unwrap(); + let address = abstract_name(b"listener"); + let listener = fixture.listener(&address, 1); + for socket in [unbound, bound, listener] { + assert_eq!( + shutdown(&fixture.process, socket, ShutdownMode::Both).unwrap(), + Ok(()) + ); + } + let client = fixture.create(SocketType::Stream); + assert_eq!( + fixture.connect(client, &address).unwrap(), + Err(UnixSocketError::ConnectionRefused) + ); +} + +#[test] +fn vsock_connects_and_addressed_sends_fail_by_the_sockets_state() { + let fixture = Fixture::new(); + let address = UnixSocketAddress::Vsock { + cid: VMADDR_CID_HOST, + port: 1024, + }; + let unbound = fixture.create(SocketType::Stream); + let bound = fixture.create(SocketType::Stream); + fixture.bind(bound, &vsock(1025)).unwrap(); + for socket in [unbound, bound] { + assert_eq!( + fixture.connect(socket, &address).unwrap(), + Err(UnixSocketError::Unreachable) + ); + assert_eq!( + fixture.send_to(socket, &address, b"x").unwrap(), + Err(UnixSocketError::Unsupported) + ); + } + let listener = fixture.listener(&vsock(1026), 1); + assert_eq!( + fixture.connect(listener, &address).unwrap(), + Err(UnixSocketError::InvalidArgument) + ); + assert_eq!( + fixture.send_to(listener, &address, b"x").unwrap(), + Err(UnixSocketError::Unsupported) + ); + + // A peer's shutdown does not count; the socket's own does. + let (host, guest) = fixture.host_connection(1024); + shutdown(&fixture.process, host, ShutdownMode::Write) + .unwrap() + .unwrap(); + assert_eq!( + fixture.connect(guest, &address).unwrap(), + Err(UnixSocketError::AlreadyConnected) + ); + assert_eq!( + fixture.connect(host, &address).unwrap(), + Err(UnixSocketError::InvalidArgument) + ); + for socket in [guest, host] { + assert_eq!( + fixture.send_to(socket, &address, b"x").unwrap(), + Err(UnixSocketError::AlreadyConnected) + ); + } +} + +fn host_port(port: u32) -> UnixSocketAddress { + UnixSocketAddress::Vsock { + cid: VMADDR_CID_HOST, + port, + } +} + +fn host_name(port: u32) -> UnixSocketName { + UnixSocketName::Vsock { + cid: VMADDR_CID_HOST, + port, + } +} + +impl Fixture { + /// The broker's host connector, with up to `backlog` connects waiting + /// beyond the first. + fn host_connector(&self, backlog: u32) -> ObjectHandle { + let connector = self.create(SocketType::Stream); + listen_for_vsock_host(&self.process, connector, backlog) + .unwrap() + .unwrap(); + connector + } + + fn connect_vsock( + &self, + handle: ObjectHandle, + address: &UnixSocketAddress, + ) -> UnixSocketResult { + connect(&self.process, handle, address, USER).unwrap() + } + + /// Starts `client`'s connect to the host's port `port`, and returns the + /// connector's socket for it. + fn connect_to_host( + &self, + connector: ObjectHandle, + client: ObjectHandle, + port: u32, + ) -> ObjectHandle { + assert_eq!( + self.connect_vsock(client, &host_port(port)), + Ok(SocketConnectionStatus::Connecting) + ); + self.accept(connector).unwrap().unwrap() + } + + fn complete(&self, host: ObjectHandle, result: UnixSocketResult<()>) -> UnixSocketResult<()> { + complete_vsock_host_connect(&self.process, host, result).unwrap() + } + + fn take_error(&self, handle: ObjectHandle) -> Option { + take_error(&self.process, handle).unwrap() + } + + fn name(&self, handle: ObjectHandle, peer: bool) -> UnixSocketResult { + name(&self.process, handle, peer).unwrap() + } +} + +#[test] +fn vsock_connects_reach_only_the_host_through_its_connector() { + let fixture = Fixture::new(); + let client = fixture.create(SocketType::Stream); + assert_eq!( + fixture.connect_vsock(client, &host_port(1024)), + Err(UnixSocketError::Unreachable) + ); + // Not bound by the attempt. + assert_eq!(fixture.name(client, false), Ok(UnixSocketName::Unnamed)); + + let connector = fixture.host_connector(8); + for cid in [VMADDR_CID_ANY, 1, 3] { + let address = UnixSocketAddress::Vsock { cid, port: 1024 }; + assert_eq!( + fixture.connect_vsock(client, &address), + Err(UnixSocketError::Unreachable) + ); + } + assert_eq!(fixture.name(client, false), Ok(UnixSocketName::Unnamed)); + // Nor do sockets of other types, or with Unix names. + for socket_type in [SocketType::Datagram, SocketType::SeqPacket] { + let socket = fixture.create(socket_type); + assert_eq!( + fixture.connect_vsock(socket, &host_port(1024)), + Err(UnixSocketError::Unreachable) + ); + } + let named = fixture.create(SocketType::Stream); + fixture.bind(named, &abstract_name(b"unix")).unwrap(); + assert_eq!( + fixture.connect_vsock(named, &host_port(1024)), + Err(UnixSocketError::Unreachable) + ); + assert_eq!(fixture.accept(connector), Err(BrokerError::WouldBlock)); + + // One connector per broker, which no guest binds or connects to. + let other = fixture.create(SocketType::Stream); + assert_eq!( + listen_for_vsock_host(&fixture.process, other, 8).unwrap(), + Err(UnixSocketError::AddressInUse) + ); + assert_eq!( + fixture.bind(connector, &vsock(1024)), + Err(UnixSocketError::InvalidArgument) + ); + assert_eq!( + fixture.connect_vsock(connector, &host_port(1024)), + Err(UnixSocketError::InvalidArgument) + ); + let bound = fixture.listener(&vsock(1024), 1); + assert_eq!( + listen_for_vsock_host(&fixture.process, bound, 8).unwrap(), + Err(UnixSocketError::InvalidArgument) + ); +} + +#[test] +fn vsock_connects_wait_for_the_host_from_an_ephemeral_or_bound_port() { + let fixture = Fixture::new(); + let connector = fixture.host_connector(8); + let client = fixture.create(SocketType::Stream); + let host = fixture.connect_to_host(connector, client, 5000); + let Ok(UnixSocketName::Vsock { + cid: VMADDR_CID_ANY, + port, + }) = fixture.name(client, false) + else { + panic!("not bound to a guest port"); + }; + assert!(port >= FIRST_EPHEMERAL_VSOCK_PORT); + // The connector's socket names both ends. + assert_eq!(fixture.name(host, false), Ok(host_name(5000))); + assert_eq!(fixture.name(host, true), Ok(guest_vsock_name(port))); + + // Connecting: no peer, no data, no readiness, nor another connect. + assert_eq!(fixture.readiness(client), ReadinessFlags::default()); + assert_eq!( + fixture.name(client, true), + Err(UnixSocketError::NotConnected) + ); + assert_eq!( + fixture.send(client, b"early").unwrap(), + Err(UnixSocketError::NotConnected) + ); + assert_eq!( + fixture.receive(client, 8).unwrap(), + Err(UnixSocketError::NotConnected) + ); + assert_eq!( + shutdown(&fixture.process, client, ShutdownMode::Write).unwrap(), + Err(UnixSocketError::NotConnected) + ); + assert_eq!( + listen(&fixture.process, client, 1).unwrap(), + Err(UnixSocketError::InvalidArgument) + ); + for address in [host_port(5000), host_port(5001), vsock(1)] { + assert_eq!( + fixture.connect_vsock(client, &address), + Err(UnixSocketError::AlreadyInProgress) + ); + } + assert_eq!(fixture.take_error(client), None); + + fixture.take_republished(); + assert_eq!(fixture.complete(host, Ok(())), Ok(())); + assert!(fixture.take_republished().contains(&client)); + assert!(fixture.readiness(client).contains(ReadinessFlags::WRITE)); + assert_eq!(fixture.name(client, true), Ok(host_name(5000))); + assert_eq!(fixture.send(client, b"ping").unwrap(), Ok(4)); + assert_eq!(fixture.receive_data(host, 8), b"ping"); + assert_eq!(fixture.send(host, b"pong").unwrap(), Ok(4)); + assert_eq!(fixture.receive_data(client, 8), b"pong"); + assert_eq!( + fixture.connect_vsock(client, &host_port(5000)), + Err(UnixSocketError::AlreadyConnected) + ); + // Completed once. + assert_eq!( + fixture.complete(host, Err(UnixSocketError::ConnectionReset)), + Err(UnixSocketError::NotConnected) + ); + + // A bound socket connects from its port. + let bound = fixture.create(SocketType::Stream); + fixture.bind(bound, &vsock(1024)).unwrap(); + let host = fixture.connect_to_host(connector, bound, 5000); + assert_eq!(fixture.name(host, true), Ok(guest_vsock_name(1024))); + assert_eq!(fixture.name(bound, false), Ok(guest_vsock_name(1024))); +} + +#[test] +fn failed_vsock_connects_report_their_error_once() { + let fixture = Fixture::new(); + let connector = fixture.host_connector(8); + let client = fixture.create(SocketType::Stream); + let host = fixture.connect_to_host(connector, client, 5000); + let name = fixture.name(client, false); + fixture.take_republished(); + assert_eq!( + fixture.complete(host, Err(UnixSocketError::ConnectionReset)), + Ok(()) + ); + assert!(fixture.take_republished().contains(&client)); + // Writable, with its error, and not hung up, as in Linux. + assert_eq!( + fixture.readiness(client), + ReadinessFlags::ERROR | ReadinessFlags::WRITE + ); + assert_eq!( + fixture.take_error(client), + Some(UnixSocketError::ConnectionReset) + ); + assert_eq!(fixture.take_error(client), None); + assert_eq!(fixture.readiness(client), ReadinessFlags::WRITE); + // Unconnected, and still bound; the connector's socket no longer reaches + // it. + assert_eq!(fixture.name(client, false), name); + assert_eq!( + fixture.name(client, true), + Err(UnixSocketError::NotConnected) + ); + assert_eq!(fixture.name(host, true), name); + fixture.close(host); + assert_eq!(fixture.take_error(client), None); + + // It connects again; a connect reports a pending error first. + let host = fixture.connect_to_host(connector, client, 5000); + assert_eq!(fixture.name(host, true), name); + assert_eq!( + fixture.complete(host, Err(UnixSocketError::TimedOut)), + Ok(()) + ); + assert_eq!( + fixture.connect_vsock(client, &host_port(5000)), + Err(UnixSocketError::TimedOut) + ); + assert_eq!(fixture.take_error(client), None); + let host = fixture.connect_to_host(connector, client, 5000); + assert_eq!(fixture.complete(host, Ok(())), Ok(())); + assert_eq!(fixture.send(client, b"x").unwrap(), Ok(1)); +} + +#[test] +fn closing_either_end_ends_a_vsock_connect() { + let fixture = Fixture::new(); + let connector = fixture.host_connector(8); + // The guest's close hangs up the connector's socket. + let client = fixture.create(SocketType::Stream); + let host = fixture.connect_to_host(connector, client, 5000); + fixture.close(client); + assert!(fixture.readiness(host).contains(ReadinessFlags::CLOSED)); + assert_eq!( + fixture.complete(host, Ok(())), + Err(UnixSocketError::NotConnected) + ); + fixture.close(host); + + // The connector's socket's close fails the connect. + let client = fixture.create(SocketType::Stream); + let host = fixture.connect_to_host(connector, client, 5000); + fixture.close(host); + assert!(fixture.readiness(client).contains(ReadinessFlags::ERROR)); + assert_eq!( + fixture.take_error(client), + Some(UnixSocketError::ConnectionReset) + ); + + // So does the connector's, before it accepts; then no connect reaches + // the host. + assert_eq!( + fixture.connect_vsock(client, &host_port(5000)), + Ok(SocketConnectionStatus::Connecting) + ); + fixture.close(connector); + assert_eq!( + fixture.connect_vsock(client, &host_port(5000)), + Err(UnixSocketError::ConnectionReset) + ); + assert_eq!( + fixture.connect_vsock(client, &host_port(5000)), + Err(UnixSocketError::Unreachable) + ); + assert_eq!(fixture.queued(), (0, 0)); +} + +#[test] +fn full_host_connectors_refuse_connects_at_once() { + let fixture = Fixture::new(); + let connector = fixture.host_connector(0); + let first = fixture.create(SocketType::Stream); + assert_eq!( + fixture.connect_vsock(first, &host_port(5000)), + Ok(SocketConnectionStatus::Connecting) + ); + let second = fixture.create(SocketType::Stream); + assert_eq!( + connect(&fixture.process, second, &host_port(5000), USER), + Err(BrokerError::ResourceExhausted) + ); + assert_eq!( + fixture.readiness(second) & ReadinessFlags::ERROR, + ReadinessFlags::default() + ); + fixture.accept(connector).unwrap().unwrap(); + let host = fixture.connect_to_host(connector, second, 5000); + assert_eq!(fixture.complete(host, Ok(())), Ok(())); +} + +#[test] +fn unconnected_vsock_sockets_are_writable_and_unix_ones_hung_up() { + let fixture = Fixture::new(); + let bound = fixture.create(SocketType::Stream); + fixture.bind(bound, &vsock(1024)).unwrap(); + assert_eq!(fixture.readiness(bound), ReadinessFlags::WRITE); + // So is one created as a vsock socket, before it has a name, and after a + // connect that named none. + let created = create_vsock( + &fixture.process, + FileOpenFlags::NONE, + &fixture.readiness_sink, + ) + .unwrap(); + assert_eq!(fixture.readiness(created), ReadinessFlags::WRITE); + assert_eq!( + fixture.connect_vsock(created, &host_port(1024)), + Err(UnixSocketError::Unreachable) + ); + assert_eq!(fixture.name(created, false), Ok(UnixSocketName::Unnamed)); + assert_eq!(fixture.readiness(created), ReadinessFlags::WRITE); + // Unix sockets, named or not, are hung up until they connect. + let named = fixture.create(SocketType::Stream); + fixture.bind(named, &abstract_name(b"unix")).unwrap(); + let unnamed = fixture.create(SocketType::Stream); + for socket in [named, unnamed] { + assert_eq!( + fixture.readiness(socket), + ReadinessFlags::WRITE | ReadinessFlags::CLOSED + ); + } +} + +#[test] +fn cancelled_vsock_connects_leave_the_socket_bound_and_hang_up_the_connector() { + let fixture = Fixture::new(); + let connector = fixture.host_connector(8); + let client = fixture.create(SocketType::Stream); + let host = fixture.connect_to_host(connector, client, 5000); + let name = fixture.name(client, false); + fixture.take_republished(); + assert_eq!( + cancel_vsock_connect(&fixture.process, client).unwrap(), + Ok(()) + ); + assert!(fixture.take_republished().contains(&client)); + // Unconnected, still bound, with no error. + assert_eq!(fixture.readiness(client), ReadinessFlags::WRITE); + assert_eq!(fixture.name(client, false), name); + assert_eq!( + fixture.name(client, true), + Err(UnixSocketError::NotConnected) + ); + assert_eq!(fixture.take_error(client), None); + // The connector's socket sees the hang-up; its completion is too late, + // and changes nothing. + assert!(fixture.readiness(host).contains(ReadinessFlags::CLOSED)); + for result in [Ok(()), Err(UnixSocketError::ConnectionReset)] { + assert_eq!( + fixture.complete(host, result), + Err(UnixSocketError::NotConnected) + ); + } + assert_eq!(fixture.readiness(client), ReadinessFlags::WRITE); + assert_eq!(fixture.take_error(client), None); + fixture.close(host); + // Nothing more to cancel. + assert_eq!( + cancel_vsock_connect(&fixture.process, client).unwrap(), + Ok(()) + ); + + // It connects again, from its port. + let host = fixture.connect_to_host(connector, client, 5000); + assert_eq!(fixture.name(host, true), name); + assert_eq!(fixture.complete(host, Ok(())), Ok(())); + // A completed connect is not cancelled; a failed one reports its error. + assert_eq!( + cancel_vsock_connect(&fixture.process, client).unwrap(), + Err(UnixSocketError::AlreadyConnected) + ); + assert_eq!(fixture.send(client, b"x").unwrap(), Ok(1)); + // As a check reports them: a refusal stays pending, a timeout does not. + let failed = fixture.create(SocketType::Stream); + for error in [UnixSocketError::ConnectionReset, UnixSocketError::TimedOut] { + let host = fixture.connect_to_host(connector, failed, 5000); + assert_eq!(fixture.complete(host, Err(error)), Ok(())); + assert_eq!( + cancel_vsock_connect(&fixture.process, failed).unwrap(), + Err(error) + ); + let pending = (error == UnixSocketError::ConnectionReset).then_some(error); + assert_eq!(fixture.take_error(failed), pending); + } +} + +#[test] +fn checks_report_vsock_connects_as_a_blocking_wait_sees_them() { + let fixture = Fixture::new(); + let connector = fixture.host_connector(8); + let client = fixture.create(SocketType::Stream); + let check = |handle| check_connect(&fixture.process, handle).unwrap(); + assert_eq!(check(client), Ok(SocketConnectionStatus::Unconnected)); + let host = fixture.connect_to_host(connector, client, 5000); + assert_eq!(check(client), Ok(SocketConnectionStatus::Connecting)); + assert_eq!( + fixture.complete(host, Err(UnixSocketError::ConnectionReset)), + Ok(()) + ); + // A refusal stays pending, for `SO_ERROR`, as in Linux. + for _ in 0..2 { + assert_eq!(check(client), Err(UnixSocketError::ConnectionReset)); + assert_eq!( + fixture.readiness(client), + ReadinessFlags::ERROR | ReadinessFlags::WRITE + ); + } + assert_eq!( + fixture.take_error(client), + Some(UnixSocketError::ConnectionReset) + ); + assert_eq!(check(client), Ok(SocketConnectionStatus::Unconnected)); + + // Linux's blocking connect times out itself, leaving no error. + let host = fixture.connect_to_host(connector, client, 5000); + assert_eq!( + fixture.complete(host, Err(UnixSocketError::TimedOut)), + Ok(()) + ); + fixture.take_republished(); + assert_eq!(check(client), Err(UnixSocketError::TimedOut)); + assert!(fixture.take_republished().contains(&client)); + assert_eq!(fixture.readiness(client), ReadinessFlags::WRITE); + assert_eq!(fixture.take_error(client), None); + + let host = fixture.connect_to_host(connector, client, 5000); + assert_eq!(fixture.complete(host, Ok(())), Ok(())); + assert_eq!(check(client), Ok(SocketConnectionStatus::Connected)); + assert_eq!(check(connector), Err(UnixSocketError::InvalidArgument)); +} + +#[test] +fn stale_hang_ups_leave_cancelled_sockets_and_their_next_connects_alone() { + let fixture = Fixture::new(); + let connector = fixture.host_connector(8); + let client = fixture.create(SocketType::Stream); + let stale_host = fixture.connect_to_host(connector, client, 5000); + let guest = fixture.socket(client).upgrade().unwrap(); + let stale = fixture.socket(stale_host).upgrade().unwrap(); + assert_eq!( + cancel_vsock_connect(&fixture.process, client).unwrap(), + Ok(()) + ); + // The connector's socket closing, having read its peer before the + // cancel, hangs up afterwards: the cancelled socket stays unconnected. + guest.hang_up(&stale); + assert_eq!(fixture.readiness(client), ReadinessFlags::WRITE); + assert_eq!(fixture.take_error(client), None); + + // Nor does it fail the next attempt, or end its connection. + let host = fixture.connect_to_host(connector, client, 5000); + guest.hang_up(&stale); + assert_eq!(fixture.readiness(client), ReadinessFlags::default()); + assert_eq!(fixture.complete(host, Ok(())), Ok(())); + guest.hang_up(&stale); + fixture.close(stale_host); + drop(stale); + assert_eq!(fixture.readiness(client), ReadinessFlags::WRITE); + assert_eq!(fixture.send(client, b"x").unwrap(), Ok(1)); + assert_eq!(fixture.receive_data(host, 8), b"x"); + assert_eq!(fixture.send(host, b"y").unwrap(), Ok(1)); + assert_eq!(fixture.receive_data(client, 8), b"y"); + assert_eq!(fixture.readiness(client), ReadinessFlags::WRITE); + assert_eq!(fixture.readiness(host), ReadinessFlags::WRITE); +} + +#[test] +fn cancels_racing_the_connectors_completion_and_close_leave_the_socket_usable() { + let fixture = Fixture::new(); + let connector = fixture.host_connector(8); + for round in 0..2000 { + let client = fixture.create(SocketType::Stream); + let host = fixture.connect_to_host(connector, client, 5000); + let barrier = std::sync::Barrier::new(2); + let cancelled = std::thread::scope(|scope| { + scope.spawn(|| { + barrier.wait(); + let result = if round % 2 == 0 { + Ok(()) + } else { + Err(UnixSocketError::ConnectionReset) + }; + let _ = fixture.complete(host, result); + fixture.close(host); + }); + barrier.wait(); + cancel_vsock_connect(&fixture.process, client).unwrap() + }); + match cancelled { + // The connection completed, and then its peer closed. + Err(UnixSocketError::AlreadyConnected) => assert_eq!( + fixture.readiness(client), + ReadinessFlags::READ + | ReadinessFlags::WRITE + | ReadinessFlags::HANGUP + | ReadinessFlags::CLOSED, + "round {round}" + ), + Ok(()) | Err(UnixSocketError::ConnectionReset) => { + // A failure stays pending; a cancel leaves no error. + let error = cancelled.err(); + let readiness = if error.is_some() { + ReadinessFlags::ERROR | ReadinessFlags::WRITE + } else { + ReadinessFlags::WRITE + }; + assert_eq!(fixture.readiness(client), readiness, "round {round}"); + assert_eq!(fixture.take_error(client), error); + // It connects again, and carries data both ways. + let host = fixture.connect_to_host(connector, client, 5000); + assert_eq!(fixture.complete(host, Ok(())), Ok(())); + assert_eq!(fixture.send(client, b"x").unwrap(), Ok(1)); + assert_eq!(fixture.receive_data(host, 8), b"x"); + assert_eq!(fixture.send(host, b"y").unwrap(), Ok(1)); + assert_eq!(fixture.receive_data(client, 8), b"y"); + assert_eq!(fixture.readiness(client), ReadinessFlags::WRITE); + fixture.close(host); + } + Err(error) => panic!("round {round}: {error:?}"), + } + fixture.close(client); + } +} + +/// How a test ends a vsock connect without connecting. +#[derive(Clone, Copy, Debug)] +enum ConnectEnd { + Failed, + Cancelled, + ConnectorSocketClosed, +} + +#[test] +fn data_queued_while_connecting_does_not_reach_a_later_connection() { + let fixture = Fixture::new(); + let connector = fixture.host_connector(8); + for end in [ + ConnectEnd::Failed, + ConnectEnd::Cancelled, + ConnectEnd::ConnectorSocketClosed, + ] { + let client = fixture.create(SocketType::Stream); + let host = fixture.connect_to_host(connector, client, 5000); + assert_eq!(fixture.send(host, b"stale").unwrap(), Ok(5), "{end:?}"); + assert_ne!(fixture.queued(), (0, 0), "{end:?}"); + match end { + ConnectEnd::Failed => assert_eq!( + fixture.complete(host, Err(UnixSocketError::ConnectionReset)), + Ok(()) + ), + ConnectEnd::Cancelled => assert_eq!( + cancel_vsock_connect(&fixture.process, client).unwrap(), + Ok(()) + ), + ConnectEnd::ConnectorSocketClosed => fixture.close(host), + } + // Discarded, and its charge refunded. + assert_eq!(fixture.queued(), (0, 0), "{end:?}"); + fixture.take_error(client); + + let host = fixture.connect_to_host(connector, client, 5000); + assert_eq!(fixture.complete(host, Ok(())), Ok(()), "{end:?}"); + assert_eq!(fixture.send(host, b"new").unwrap(), Ok(3), "{end:?}"); + assert_eq!(fixture.receive_data(client, 16), b"new", "{end:?}"); + assert_eq!( + fixture.receive(client, 16), + Err(BrokerError::WouldBlock), + "{end:?}" + ); + } +} + +#[test] +fn connecting_sockets_receive_data_queued_for_them_once_connected() { + let fixture = Fixture::new(); + let connector = fixture.host_connector(8); + let client = fixture.create(SocketType::Stream); + let host = fixture.connect_to_host(connector, client, 5000); + assert_eq!(fixture.send(host, b"early").unwrap(), Ok(5)); + // Not readable while connecting. + assert_eq!(fixture.readiness(client), ReadinessFlags::default()); + assert_eq!( + fixture.receive(client, 16).unwrap(), + Err(UnixSocketError::NotConnected) + ); + assert_eq!(fixture.complete(host, Ok(())), Ok(())); + assert_eq!( + fixture.readiness(client), + ReadinessFlags::READ | ReadinessFlags::WRITE + ); + assert_eq!(fixture.send(host, b"late").unwrap(), Ok(4)); + assert_eq!(fixture.receive_data(client, 16), b"earlylate"); +} + +#[test] +fn failed_completions_end_the_connectors_socket() { + let fixture = Fixture::new(); + let connector = fixture.host_connector(8); + let client = fixture.create(SocketType::Stream); + let host = fixture.connect_to_host(connector, client, 5000); + fixture.take_republished(); + assert_eq!( + fixture.complete(host, Err(UnixSocketError::ConnectionReset)), + Ok(()) + ); + assert!(fixture.take_republished().contains(&host)); + assert_eq!( + fixture.send(host, b"x").unwrap(), + Err(UnixSocketError::BrokenPipe) + ); + assert_eq!( + fixture.receive(host, 8).unwrap(), + Ok(Message { + source: Some(fixture.name(client, false).unwrap()), + ..Message::default() + }) + ); + assert!( + fixture + .readiness(host) + .contains(ReadinessFlags::READ | ReadinessFlags::HANGUP | ReadinessFlags::CLOSED) + ); +} diff --git a/litebox_broker_host/src/lib.rs b/litebox_broker_host/src/lib.rs index b231576135..a6ed20fa29 100644 --- a/litebox_broker_host/src/lib.rs +++ b/litebox_broker_host/src/lib.rs @@ -77,10 +77,11 @@ use litebox_broker_protocol::timer::{ CreateTimerResponse, GetTimerResponse, ReadTimerResponse, SetTimerResponse, }; use litebox_broker_protocol::unix_socket::{ - AcceptUnixSocketResponse, CreateUnixSocketPairResponse, CreateUnixSocketResponse, - GetUnixSocketNameResponse, GetUnixSocketOptionResponse, MAX_ENCODED_UNIX_SOCKET_ADDRESS_SIZE, + AcceptUnixSocketResponse, CheckUnixSocketConnectResponse, ConnectUnixSocketResponse, + CreateUnixSocketPairResponse, CreateUnixSocketResponse, GetUnixSocketNameResponse, + GetUnixSocketOptionResponse, MAX_ENCODED_UNIX_SOCKET_ADDRESS_SIZE, MAX_ENCODED_UNIX_SOCKET_NAME_SIZE, MAX_UNIX_SOCKET_TRANSFER_SIZE, ReceiveUnixSocketResponse, - SendUnixSocketResponse, UnixSocketAddress, UnixSocketError, + SendUnixSocketResponse, TakeUnixSocketErrorResponse, UnixSocketAddress, UnixSocketError, }; use litebox_broker_protocol::{BROKER_PROTOCOL_VERSION, ObjectHandle, ProcessId, RequestId}; use litebox_broker_transport::channel::{HostReceive, HostSetupChannel, PeerCredential}; @@ -1613,6 +1614,12 @@ fn handle_unix_socket_request( handle, })) } + UnixSocketRequest::CreateVsock(request) => { + let handle = unix_socket::create_vsock(process, request.flags, readiness_sink)?; + Ok(UnixSocketResponse::CreateVsock(CreateUnixSocketResponse { + handle, + })) + } UnixSocketRequest::CreatePair(request) => { let (first, second) = unix_socket::create_pair( process, @@ -1651,7 +1658,9 @@ fn handle_unix_socket_request( }; Ok( unix_socket::connect(process, request.handle, &address, request.user)? - .map_or_else(UnixSocketResponse::Failed, |()| UnixSocketResponse::Connect), + .map_or_else(UnixSocketResponse::Failed, |status| { + UnixSocketResponse::Connect(ConnectUnixSocketResponse { status }) + }), ) } UnixSocketRequest::Accept(request) => { @@ -1789,6 +1798,25 @@ fn handle_unix_socket_request( value, })) } + UnixSocketRequest::TakeError(request) => { + let error = unix_socket::take_error(process, request.handle)?; + Ok(UnixSocketResponse::TakeError(TakeUnixSocketErrorResponse { + error, + })) + } + UnixSocketRequest::CheckConnect(request) => Ok(unix_socket::check_connect( + process, + request.handle, + )? + .map_or_else(UnixSocketResponse::Failed, |status| { + UnixSocketResponse::CheckConnect(CheckUnixSocketConnectResponse { status }) + })), + UnixSocketRequest::CancelConnect(request) => { + Ok(unix_socket::cancel_vsock_connect(process, request.handle)? + .map_or_else(UnixSocketResponse::Failed, |()| { + UnixSocketResponse::CancelConnect + })) + } } } diff --git a/litebox_broker_local/src/unix_socket.rs b/litebox_broker_local/src/unix_socket.rs index da9fd53029..4bb783ed47 100644 --- a/litebox_broker_local/src/unix_socket.rs +++ b/litebox_broker_local/src/unix_socket.rs @@ -9,15 +9,16 @@ use litebox_broker_protocol::message::{ }; use litebox_broker_protocol::shared_buffer::SharedBufferSequence; use litebox_broker_protocol::socket::{ - ShutdownMode, SocketOptionName, SocketOptionValue, SocketType, + ShutdownMode, SocketConnectionStatus, SocketOptionName, SocketOptionValue, SocketType, }; use litebox_broker_protocol::unix_socket::{ - AcceptUnixSocketRequest, BindUnixSocketRequest, ConnectUnixSocketRequest, - CreateUnixSocketPairResponse, CreateUnixSocketRequest, GetUnixSocketNameRequest, + AcceptUnixSocketRequest, BindUnixSocketRequest, CancelUnixSocketConnectRequest, + CheckUnixSocketConnectRequest, ConnectUnixSocketRequest, CreateUnixSocketPairResponse, + CreateUnixSocketRequest, CreateVsockSocketRequest, GetUnixSocketNameRequest, GetUnixSocketOptionRequest, ListenUnixSocketRequest, MAX_ENCODED_UNIX_SOCKET_ADDRESS_SIZE, MAX_ENCODED_UNIX_SOCKET_NAME_SIZE, MAX_UNIX_SOCKET_TRANSFER_SIZE, ReceiveUnixSocketRequest, ReceiveUnixSocketResponse, SendUnixSocketRequest, SetUnixSocketOptionRequest, - ShutdownUnixSocketRequest, UnixSocketError, UnixSocketName, + ShutdownUnixSocketRequest, TakeUnixSocketErrorRequest, UnixSocketError, UnixSocketName, }; use litebox_broker_transport::channel::LocalCallChannel; use litebox_platform::buffer::{ReadBuffer, WriteBuffer}; @@ -46,6 +47,24 @@ impl BrokerLocal { } } + /// Creates a broker-owned vsock stream socket (see + /// [`CreateVsockSocketRequest`]). + /// + /// # Panics + /// + /// Panics if the broker returns a response for a different operation. + pub fn create_vsock_socket( + &self, + flags: FileOpenFlags, + ) -> Result { + match self.request_unix_socket(UnixSocketRequest::CreateVsock( + CreateVsockSocketRequest { flags }, + ))? { + UnixSocketResponse::CreateVsock(response) => Ok(response.handle), + response => panic!("broker returned unexpected Unix socket response: {response:?}"), + } + } + /// Creates a pair of connected broker-owned Unix sockets. /// /// # Panics @@ -114,6 +133,9 @@ impl BrokerLocal { /// Connects a Unix socket to an encoded address staged in `buffer`. /// + /// Returns whether it is connected, or connecting (see + /// [`ConnectUnixSocketResponse`](litebox_broker_protocol::unix_socket::ConnectUnixSocketResponse)). + /// /// # Panics /// /// Panics if `buffer` does not match `address` or the broker returns a @@ -124,14 +146,14 @@ impl BrokerLocal { buffer: SharedBufferSequence, address: &[u8], user: FileUser, - ) -> Result, Channel::Error> { + ) -> Result, Channel::Error> { self.stage_unix_socket_buffer(buffer, address, MAX_ENCODED_UNIX_SOCKET_ADDRESS_SIZE)?; match self.request_unix_socket(UnixSocketRequest::Connect(ConnectUnixSocketRequest { handle, address: buffer, user, }))? { - UnixSocketResponse::Connect => Ok(Ok(())), + UnixSocketResponse::Connect(response) => Ok(Ok(response.status)), UnixSocketResponse::Failed(error) => Ok(Err(error)), response => panic!("broker returned unexpected Unix socket response: {response:?}"), } @@ -369,6 +391,61 @@ impl BrokerLocal { Ok(response.value) } + /// Takes the pending error of a Unix socket. + /// + /// # Panics + /// + /// Panics if the broker returns a response for a different operation. + pub fn take_unix_socket_error( + &self, + handle: ObjectHandle, + ) -> Result, Channel::Error> { + match self.request_unix_socket(UnixSocketRequest::TakeError( + TakeUnixSocketErrorRequest { handle }, + ))? { + UnixSocketResponse::TakeError(response) => Ok(response.error), + response => panic!("broker returned unexpected Unix socket response: {response:?}"), + } + } + + /// Reports the outcome of a Unix socket's vsock connect (see + /// [`CheckUnixSocketConnectRequest`]). + /// + /// # Panics + /// + /// Panics if the broker returns a response for a different operation. + pub fn check_unix_socket_connect( + &self, + handle: ObjectHandle, + ) -> Result, Channel::Error> { + match self.request_unix_socket(UnixSocketRequest::CheckConnect( + CheckUnixSocketConnectRequest { handle }, + ))? { + UnixSocketResponse::CheckConnect(response) => Ok(Ok(response.status)), + UnixSocketResponse::Failed(error) => Ok(Err(error)), + response => panic!("broker returned unexpected Unix socket response: {response:?}"), + } + } + + /// Cancels a Unix socket's vsock connect in progress (see + /// [`CancelUnixSocketConnectRequest`]). + /// + /// # Panics + /// + /// Panics if the broker returns a response for a different operation. + pub fn cancel_unix_socket_connect( + &self, + handle: ObjectHandle, + ) -> Result, Channel::Error> { + match self.request_unix_socket(UnixSocketRequest::CancelConnect( + CancelUnixSocketConnectRequest { handle }, + ))? { + UnixSocketResponse::CancelConnect => Ok(Ok(())), + UnixSocketResponse::Failed(error) => Ok(Err(error)), + response => panic!("broker returned unexpected Unix socket response: {response:?}"), + } + } + fn stage_unix_socket_buffer( &self, buffer: SharedBufferSequence, diff --git a/litebox_broker_protocol/src/message.rs b/litebox_broker_protocol/src/message.rs index ce2c432d53..39a6b04e8d 100644 --- a/litebox_broker_protocol/src/message.rs +++ b/litebox_broker_protocol/src/message.rs @@ -57,11 +57,14 @@ use crate::timer::{ }; use crate::unix_socket::{ AcceptUnixSocketRequest, AcceptUnixSocketResponse, BindUnixSocketRequest, - ConnectUnixSocketRequest, CreateUnixSocketPairResponse, CreateUnixSocketRequest, - CreateUnixSocketResponse, GetUnixSocketNameRequest, GetUnixSocketNameResponse, - GetUnixSocketOptionRequest, GetUnixSocketOptionResponse, ListenUnixSocketRequest, - ReceiveUnixSocketRequest, ReceiveUnixSocketResponse, SendUnixSocketRequest, - SendUnixSocketResponse, SetUnixSocketOptionRequest, ShutdownUnixSocketRequest, UnixSocketError, + CancelUnixSocketConnectRequest, CheckUnixSocketConnectRequest, CheckUnixSocketConnectResponse, + ConnectUnixSocketRequest, ConnectUnixSocketResponse, CreateUnixSocketPairResponse, + CreateUnixSocketRequest, CreateUnixSocketResponse, CreateVsockSocketRequest, + GetUnixSocketNameRequest, GetUnixSocketNameResponse, GetUnixSocketOptionRequest, + GetUnixSocketOptionResponse, ListenUnixSocketRequest, ReceiveUnixSocketRequest, + ReceiveUnixSocketResponse, SendUnixSocketRequest, SendUnixSocketResponse, + SetUnixSocketOptionRequest, ShutdownUnixSocketRequest, TakeUnixSocketErrorRequest, + TakeUnixSocketErrorResponse, UnixSocketError, }; use crate::{ObjectHandle, ProcessId, ProtocolVersion, RequestId, ThreadId}; @@ -219,11 +222,15 @@ impl BrokerOperation { | Self::UnixSocket( UnixSocketRequest::Create(_) | UnixSocketRequest::CreatePair(_) + | UnixSocketRequest::CreateVsock(_) | UnixSocketRequest::Listen(_) | UnixSocketRequest::Accept(_) | UnixSocketRequest::Shutdown(_) | UnixSocketRequest::SetOption(_) - | UnixSocketRequest::GetOption(_), + | UnixSocketRequest::GetOption(_) + | UnixSocketRequest::TakeError(_) + | UnixSocketRequest::CheckConnect(_) + | UnixSocketRequest::CancelConnect(_), ) | Self::ReadinessQueue( ReadinessQueueRequest::Create @@ -461,6 +468,8 @@ pub enum UnixSocketRequest { Create(CreateUnixSocketRequest), /// Create a pair of connected unnamed sockets. CreatePair(CreateUnixSocketRequest), + /// Create an unbound, unconnected vsock stream socket. + CreateVsock(CreateVsockSocketRequest), /// Bind a socket to a name. Bind(BindUnixSocketRequest), /// Make a bound stream or sequenced-packet socket accept connections. @@ -481,6 +490,12 @@ pub enum UnixSocketRequest { SetOption(SetUnixSocketOptionRequest), /// Read one socket option. GetOption(GetUnixSocketOptionRequest), + /// Take a socket's pending error. + TakeError(TakeUnixSocketErrorRequest), + /// Report the outcome of a socket's vsock connect. + CheckConnect(CheckUnixSocketConnectRequest), + /// Cancel a socket's vsock connect in progress. + CancelConnect(CancelUnixSocketConnectRequest), } /// Broker-owned socket object request. @@ -760,12 +775,14 @@ pub enum UnixSocketResponse { Create(CreateUnixSocketResponse), /// Pair create operation response. CreatePair(CreateUnixSocketPairResponse), + /// Vsock socket create operation response. + CreateVsock(CreateUnixSocketResponse), /// Bind operation completed. Bind, /// Listen operation completed. Listen, - /// Connect operation completed. - Connect, + /// Connect operation response. + Connect(ConnectUnixSocketResponse), /// Accept operation response. Accept(AcceptUnixSocketResponse), /// Send operation response. @@ -780,6 +797,12 @@ pub enum UnixSocketResponse { SetOption, /// Option read response. GetOption(GetUnixSocketOptionResponse), + /// Pending error response. + TakeError(TakeUnixSocketErrorResponse), + /// Check-connect response. + CheckConnect(CheckUnixSocketConnectResponse), + /// Cancel operation completed. + CancelConnect, /// The operation failed in a way the guest ABI reports. /// /// Waiting, resource, and request-validation failures use diff --git a/litebox_broker_protocol/src/unix_socket.rs b/litebox_broker_protocol/src/unix_socket.rs index 0e3fe810d2..634e712c0c 100644 --- a/litebox_broker_protocol/src/unix_socket.rs +++ b/litebox_broker_protocol/src/unix_socket.rs @@ -8,6 +8,18 @@ //! the path resolves to, or a byte string in an abstract namespace. Data moves //! through broker-owned queues, so every reference to a socket, including //! references in other processes, shares its state. +//! +//! A stream socket may instead be named by a vsock address, in a namespace of +//! its own that no path or abstract name reaches. Connections to a vsock +//! listener come from the broker on behalf of the VM's host, and a guest's +//! connect to a vsock address reaches only the host, through the broker, if +//! it has a vsock device. Such a connect completes later: its socket reports +//! [`SocketConnectionStatus::Connecting`] until then, and a failure is the +//! socket's pending error ([`TakeUnixSocketErrorRequest`]), which a blocking +//! connect's wait reads ([`CheckUnixSocketConnectRequest`]); it may be +//! cancelled meanwhile ([`CancelUnixSocketConnectRequest`]). A socket +//! created as a vsock socket ([`CreateVsockSocketRequest`]) is one even +//! before it has a name. use alloc::string::String; use alloc::vec::Vec; @@ -17,11 +29,17 @@ use thiserror::Error; use crate::ObjectHandle; use crate::fs::{FileError, FileMode, FileOpenFlags, FileUser}; use crate::shared_buffer::{SHARED_BUFFER_SLOT_SIZE, SharedBufferSequence}; -use crate::socket::{ShutdownMode, SocketOptionName, SocketOptionValue, SocketType}; +use crate::socket::{ + ShutdownMode, SocketConnectionStatus, SocketOptionName, SocketOptionValue, SocketType, +}; /// Maximum length in bytes of the guest-visible part of a Unix socket name. pub const MAX_UNIX_SOCKET_NAME_SIZE: u32 = 108; +/// Length in bytes of the encoded payload of a vsock name or address: its +/// CID and port. +const VSOCK_NAME_SIZE: u32 = 8; + /// Maximum length in bytes of an encoded [`UnixSocketName`]. pub const MAX_ENCODED_UNIX_SOCKET_NAME_SIZE: u32 = 1 + MAX_UNIX_SOCKET_NAME_SIZE; @@ -51,10 +69,22 @@ const _: () = assert!( <= crate::shared_buffer::MAX_SHARED_BUFFER_SEQUENCE_SLOTS ); const _: () = assert!(MAX_UNIX_SOCKET_PATH_SIZE <= u16::MAX as u32); +const _: () = assert!(VSOCK_NAME_SIZE <= MAX_UNIX_SOCKET_NAME_SIZE); const NAME_TAG_UNNAMED: u8 = 0; const NAME_TAG_PATH: u8 = 1; const NAME_TAG_ABSTRACT: u8 = 2; +const NAME_TAG_VSOCK: u8 = 3; + +/// `VMADDR_CID_ANY`: the CID of a vsock name a guest binds, which accepts +/// connections to any of the VM's CIDs. +pub const VMADDR_CID_ANY: u32 = u32::MAX; + +/// `VMADDR_CID_HOST`: the CID of the VM's host. +pub const VMADDR_CID_HOST: u32 = 2; + +/// `VMADDR_PORT_ANY`: binding to it binds an unused ephemeral port. +pub const VMADDR_PORT_ANY: u32 = u32::MAX; /// The name a Unix socket reports for itself or its peer. #[derive(Clone, Debug, Default, PartialEq, Eq)] @@ -66,6 +96,14 @@ pub enum UnixSocketName { Path(Vec), /// A name in the abstract namespace. Abstract(Vec), + /// A vsock address. + Vsock { + /// Context ID: [`VMADDR_CID_ANY`] for a guest's socket, + /// [`VMADDR_CID_HOST`] for the host's. + cid: u32, + /// Port. + port: u32, + }, } impl UnixSocketName { @@ -78,6 +116,7 @@ impl UnixSocketName { Self::Unnamed => (NAME_TAG_UNNAMED, &[]), Self::Path(bytes) => (NAME_TAG_PATH, bytes), Self::Abstract(bytes) => (NAME_TAG_ABSTRACT, bytes), + &Self::Vsock { cid, port } => return Some(encode_vsock(cid, port)), }; if bytes.len() > MAX_UNIX_SOCKET_NAME_SIZE as usize { return None; @@ -100,11 +139,40 @@ impl UnixSocketName { NAME_TAG_UNNAMED if bytes.is_empty() => Ok(Self::Unnamed), NAME_TAG_PATH => Ok(Self::Path(bytes.into())), NAME_TAG_ABSTRACT => Ok(Self::Abstract(bytes.into())), + NAME_TAG_VSOCK => { + let (cid, port) = decode_vsock(bytes)?; + Ok(Self::Vsock { cid, port }) + } _ => Err(UnixSocketCodecError::Invalid), } } } +fn encode_vsock(cid: u32, port: u32) -> Vec { + let mut encoded = Vec::with_capacity(1 + VSOCK_NAME_SIZE as usize); + encoded.push(NAME_TAG_VSOCK); + encoded.extend_from_slice(&cid.to_le_bytes()); + encoded.extend_from_slice(&port.to_le_bytes()); + encoded +} + +/// Decodes the payload of a vsock name or address, which is exactly its CID +/// and port. +fn decode_vsock(bytes: &[u8]) -> Result<(u32, u32), UnixSocketCodecError> { + let bytes: &[u8; VSOCK_NAME_SIZE as usize] = bytes.try_into().map_err(|_| { + if bytes.len() < VSOCK_NAME_SIZE as usize { + UnixSocketCodecError::Truncated + } else { + UnixSocketCodecError::TooLong + } + })?; + let [c0, c1, c2, c3, p0, p1, p2, p3] = *bytes; + Ok(( + u32::from_le_bytes([c0, c1, c2, c3]), + u32::from_le_bytes([p0, p1, p2, p3]), + )) +} + /// An address that binds, connects, or sends to a Unix socket. #[derive(Clone, Debug, PartialEq, Eq)] pub enum UnixSocketAddress { @@ -117,15 +185,24 @@ pub enum UnixSocketAddress { }, /// A name in the abstract namespace. Abstract(Vec), + /// A vsock address, which only a stream socket binds. + Vsock { + /// Context ID, which a guest binds only as [`VMADDR_CID_ANY`]. + cid: u32, + /// Port, or [`VMADDR_PORT_ANY`] to bind an unused ephemeral one. + port: u32, + }, } impl UnixSocketAddress { - /// Returns the name a socket bound to this address reports. + /// Returns the name a socket bound to this address reports, but for + /// [`VMADDR_PORT_ANY`], which binds another port. #[must_use] pub fn name(&self) -> UnixSocketName { match self { Self::Path { name, .. } => UnixSocketName::Path(name.clone()), Self::Abstract(bytes) => UnixSocketName::Abstract(bytes.clone()), + &Self::Vsock { cid, port } => UnixSocketName::Vsock { cid, port }, } } @@ -150,6 +227,7 @@ impl UnixSocketAddress { Some(encoded) } Self::Abstract(bytes) => UnixSocketName::Abstract(bytes.clone()).encode(), + &Self::Vsock { cid, port } => Some(encode_vsock(cid, port)), } } @@ -183,6 +261,10 @@ impl UnixSocketAddress { UnixSocketName::Abstract(bytes) => Ok(Self::Abstract(bytes)), _ => Err(UnixSocketCodecError::Invalid), }, + Some((&NAME_TAG_VSOCK, bytes)) => { + let (cid, port) = decode_vsock(bytes)?; + Ok(Self::Vsock { cid, port }) + } Some(_) => Err(UnixSocketCodecError::Invalid), None => Err(UnixSocketCodecError::Truncated), } @@ -226,6 +308,14 @@ pub enum UnixSocketError { NotPermitted, #[error("connection was reset")] ConnectionReset, + #[error("address is not available to bind")] + AddressNotAvailable, + #[error("address is unreachable")] + Unreachable, + #[error("a connection attempt is already in progress")] + AlreadyInProgress, + #[error("connection attempt timed out")] + TimedOut, #[error("address lookup failed: {0}")] File(FileError), } @@ -239,7 +329,15 @@ pub struct CreateUnixSocketRequest { pub flags: FileOpenFlags, } -/// Response to a Unix socket create request. +/// Request to create an unbound, unconnected vsock stream socket, which, as +/// in Linux, is not hung up while unconnected, even before it has a name. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct CreateVsockSocketRequest { + /// Initial status flags, within [`FileOpenFlags::STATUS`]. + pub flags: FileOpenFlags, +} + +/// Response to a Unix socket or vsock socket create request. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct CreateUnixSocketResponse { /// Handle for the new socket. @@ -293,6 +391,15 @@ pub struct ConnectUnixSocketRequest { pub user: FileUser, } +/// Response to a connect request. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ConnectUnixSocketResponse { + /// [`SocketConnectionStatus::Connected`], or + /// [`SocketConnectionStatus::Connecting`] for a vsock connect that + /// completes later. + pub status: SocketConnectionStatus, +} + /// Request to accept one pending connection. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct AcceptUnixSocketRequest { @@ -433,6 +540,52 @@ pub struct GetUnixSocketOptionResponse { pub value: SocketOptionValue, } +/// Request to take a socket's pending error: why its last connect failed, +/// which is reported once. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct TakeUnixSocketErrorRequest { + /// Socket handle. + pub handle: ObjectHandle, +} + +/// Request for the outcome of a socket's vsock connect, as a blocking +/// connect waiting for it sees it: whether it is still in progress, or +/// connected, or the error it failed with, which stays pending unless the +/// connect timed out. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct CheckUnixSocketConnectRequest { + /// Socket handle. + pub handle: ObjectHandle, +} + +/// Response to a check-connect request. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct CheckUnixSocketConnectResponse { + /// [`SocketConnectionStatus::Connecting`], + /// [`SocketConnectionStatus::Connected`], or + /// [`SocketConnectionStatus::Unconnected`] for a socket with no connect + /// in progress and no pending error. + pub status: SocketConnectionStatus, +} + +/// Request to cancel a socket's vsock connect in progress, as an interrupted +/// blocking connect does: the socket is unconnected again, and still bound. +/// With none in progress, it reports the outcome as +/// [`CheckUnixSocketConnectRequest`] does, failing with +/// [`UnixSocketError::AlreadyConnected`] if the connect completed. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct CancelUnixSocketConnectRequest { + /// Socket handle. + pub handle: ObjectHandle, +} + +/// Response containing a socket's pending error. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct TakeUnixSocketErrorResponse { + /// Taken error, or `None` if no error was pending. + pub error: Option, +} + #[cfg(test)] mod tests { use super::*; @@ -444,8 +597,18 @@ mod tests { UnixSocketName::Path(b"./sock".into()), UnixSocketName::Abstract(b"\0x".into()), UnixSocketName::Abstract(Vec::new()), + UnixSocketName::Vsock { + cid: VMADDR_CID_HOST, + port: 0x1234_5678, + }, + UnixSocketName::Vsock { + cid: VMADDR_CID_ANY, + port: VMADDR_PORT_ANY, + }, ] { - assert_eq!(UnixSocketName::decode(&name.encode().unwrap()), Ok(name)); + let encoded = name.encode().unwrap(); + assert!(encoded.len() <= MAX_ENCODED_UNIX_SOCKET_NAME_SIZE as usize); + assert_eq!(UnixSocketName::decode(&encoded), Ok(name)); } assert_eq!( UnixSocketName::Path(alloc::vec![1; MAX_UNIX_SOCKET_NAME_SIZE as usize + 1]).encode(), @@ -469,6 +632,10 @@ mod tests { name: b"sock".into(), }, UnixSocketAddress::Abstract(b"name".into()), + UnixSocketAddress::Vsock { + cid: VMADDR_CID_ANY, + port: 1024, + }, ] { let encoded = address.encode().unwrap(); assert!(encoded.len() <= MAX_ENCODED_UNIX_SOCKET_ADDRESS_SIZE as usize); @@ -491,4 +658,42 @@ mod tests { Err(UnixSocketCodecError::Truncated) ); } + + #[test] + fn vsock_names_are_little_endian_and_exactly_sized() { + let name = UnixSocketName::Vsock { + cid: 2, + port: 0x0102_0304, + }; + let encoded = name.encode().unwrap(); + assert_eq!(encoded, [NAME_TAG_VSOCK, 2, 0, 0, 0, 4, 3, 2, 1]); + assert_eq!( + UnixSocketAddress::decode(&encoded), + Ok(UnixSocketAddress::Vsock { + cid: 2, + port: 0x0102_0304 + }) + ); + for length in 0..encoded.len() - 1 { + let truncated = &encoded[..=length]; + assert_eq!( + UnixSocketName::decode(truncated), + Err(UnixSocketCodecError::Truncated) + ); + assert_eq!( + UnixSocketAddress::decode(truncated), + Err(UnixSocketCodecError::Truncated) + ); + } + let mut long = encoded; + long.push(0); + assert_eq!( + UnixSocketName::decode(&long), + Err(UnixSocketCodecError::TooLong) + ); + assert_eq!( + UnixSocketAddress::decode(&long), + Err(UnixSocketCodecError::TooLong) + ); + } } diff --git a/litebox_broker_protocol/src/wire.rs b/litebox_broker_protocol/src/wire.rs index cd83ae5fe4..4e7a5f03f0 100644 --- a/litebox_broker_protocol/src/wire.rs +++ b/litebox_broker_protocol/src/wire.rs @@ -713,12 +713,14 @@ mod tests { }; use crate::unix_socket::{ AcceptUnixSocketRequest, AcceptUnixSocketResponse, BindUnixSocketRequest, - ConnectUnixSocketRequest, CreateUnixSocketPairResponse, CreateUnixSocketRequest, - CreateUnixSocketResponse, GetUnixSocketNameRequest, GetUnixSocketNameResponse, + CancelUnixSocketConnectRequest, CheckUnixSocketConnectRequest, + CheckUnixSocketConnectResponse, ConnectUnixSocketRequest, ConnectUnixSocketResponse, + CreateUnixSocketPairResponse, CreateUnixSocketRequest, CreateUnixSocketResponse, + CreateVsockSocketRequest, GetUnixSocketNameRequest, GetUnixSocketNameResponse, GetUnixSocketOptionRequest, GetUnixSocketOptionResponse, ListenUnixSocketRequest, ReceiveUnixSocketRequest, ReceiveUnixSocketResponse, SendUnixSocketRequest, SendUnixSocketResponse, SetUnixSocketOptionRequest, ShutdownUnixSocketRequest, - UnixSocketError, + TakeUnixSocketErrorRequest, TakeUnixSocketErrorResponse, UnixSocketError, }; use crate::{ ObjectHandle, ProcessGroupId, ProcessId, ProtocolVersion, RequestId, SessionId, ThreadId, @@ -1259,6 +1261,12 @@ mod tests { socket_type: SocketType::Stream, flags: FileOpenFlags::NONBLOCKING, })), + BrokerOperation::UnixSocket(UnixSocketRequest::CreateVsock(CreateVsockSocketRequest { + flags: FileOpenFlags::NONE, + })), + BrokerOperation::UnixSocket(UnixSocketRequest::CreateVsock(CreateVsockSocketRequest { + flags: FileOpenFlags::NONBLOCKING, + })), BrokerOperation::UnixSocket(UnixSocketRequest::CreatePair(CreateUnixSocketRequest { socket_type: SocketType::Datagram, flags: FileOpenFlags::NONE, @@ -1331,6 +1339,15 @@ mod tests { handle, name: SocketOptionName::SendTimeout, })), + BrokerOperation::UnixSocket(UnixSocketRequest::TakeError(TakeUnixSocketErrorRequest { + handle, + })), + BrokerOperation::UnixSocket(UnixSocketRequest::CheckConnect( + CheckUnixSocketConnectRequest { handle }, + )), + BrokerOperation::UnixSocket(UnixSocketRequest::CancelConnect( + CancelUnixSocketConnectRequest { handle }, + )), ]; let mut maximum_encoded_size = 0; @@ -1591,9 +1608,33 @@ mod tests { second: ObjectHandle(u64::MAX), }, )), + BrokerResult::UnixSocket(UnixSocketResponse::CreateVsock(CreateUnixSocketResponse { + handle, + })), BrokerResult::UnixSocket(UnixSocketResponse::Bind), BrokerResult::UnixSocket(UnixSocketResponse::Listen), - BrokerResult::UnixSocket(UnixSocketResponse::Connect), + BrokerResult::UnixSocket(UnixSocketResponse::Connect(ConnectUnixSocketResponse { + status: SocketConnectionStatus::Connected, + })), + BrokerResult::UnixSocket(UnixSocketResponse::Connect(ConnectUnixSocketResponse { + status: SocketConnectionStatus::Connecting, + })), + BrokerResult::UnixSocket(UnixSocketResponse::CheckConnect( + CheckUnixSocketConnectResponse { + status: SocketConnectionStatus::Unconnected, + }, + )), + BrokerResult::UnixSocket(UnixSocketResponse::CheckConnect( + CheckUnixSocketConnectResponse { + status: SocketConnectionStatus::Connecting, + }, + )), + BrokerResult::UnixSocket(UnixSocketResponse::CheckConnect( + CheckUnixSocketConnectResponse { + status: SocketConnectionStatus::Connected, + }, + )), + BrokerResult::UnixSocket(UnixSocketResponse::CancelConnect), BrokerResult::UnixSocket(UnixSocketResponse::Accept(AcceptUnixSocketResponse { handle, })), @@ -1622,6 +1663,22 @@ mod tests { BrokerResult::UnixSocket(UnixSocketResponse::Failed(UnixSocketError::AddressInUse)), BrokerResult::UnixSocket(UnixSocketResponse::Failed(UnixSocketError::NotPermitted)), BrokerResult::UnixSocket(UnixSocketResponse::Failed(UnixSocketError::ConnectionReset)), + BrokerResult::UnixSocket(UnixSocketResponse::Failed( + UnixSocketError::AddressNotAvailable, + )), + BrokerResult::UnixSocket(UnixSocketResponse::Failed(UnixSocketError::Unreachable)), + BrokerResult::UnixSocket(UnixSocketResponse::Failed( + UnixSocketError::AlreadyInProgress, + )), + BrokerResult::UnixSocket(UnixSocketResponse::TakeError(TakeUnixSocketErrorResponse { + error: Some(UnixSocketError::TimedOut), + })), + BrokerResult::UnixSocket(UnixSocketResponse::TakeError(TakeUnixSocketErrorResponse { + error: Some(UnixSocketError::File(FileError::AccessNotAllowed)), + })), + BrokerResult::UnixSocket(UnixSocketResponse::TakeError(TakeUnixSocketErrorResponse { + error: None, + })), BrokerResult::UnixSocket(UnixSocketResponse::Failed(UnixSocketError::File( FileError::NoSuchFileOrDirectory, ))), @@ -2694,6 +2751,34 @@ mod tests { decode_response(&unknown_thread_response), Err(WireError::InvalidTag) ); + // A connect completes connected or connecting, nothing else. + let mut unconnected = encode_response(BrokerResponse { + request_id: TEST_REQUEST_ID, + result: BrokerResult::UnixSocket(UnixSocketResponse::Connect( + ConnectUnixSocketResponse { + status: SocketConnectionStatus::Connecting, + }, + )), + }); + *unconnected.last_mut().unwrap() = 0; + assert_eq!(decode_response(&unconnected), Err(WireError::InvalidTag)); + // A check reports a failure as one, not as a status. + let mut failed = encode_response(BrokerResponse { + request_id: TEST_REQUEST_ID, + result: BrokerResult::UnixSocket(UnixSocketResponse::CheckConnect( + CheckUnixSocketConnectResponse { + status: SocketConnectionStatus::Connected, + }, + )), + }); + failed.pop(); + let mut status = primitive::Encoder::default(); + socket::encode_connection_status( + &mut status, + SocketConnectionStatus::Failed(SocketError::ConnectionRefused), + ); + failed.extend(status.finish()); + assert_eq!(decode_response(&failed), Err(WireError::InvalidTag)); let mut unknown_process_response = encode_response(BrokerResponse { request_id: TEST_REQUEST_ID, result: BrokerResult::Process(ProcessResponse::StartChild), diff --git a/litebox_broker_protocol/src/wire/socket.rs b/litebox_broker_protocol/src/wire/socket.rs index 7624032f1d..ae2942409d 100644 --- a/litebox_broker_protocol/src/wire/socket.rs +++ b/litebox_broker_protocol/src/wire/socket.rs @@ -512,7 +512,7 @@ pub(super) fn decode_option_value( }) } -fn encode_connection_status(encoder: &mut Encoder, status: SocketConnectionStatus) { +pub(super) fn encode_connection_status(encoder: &mut Encoder, status: SocketConnectionStatus) { match status { SocketConnectionStatus::Unconnected => encoder.u8(CONNECTION_STATUS_TAG_UNCONNECTED), SocketConnectionStatus::Connecting => encoder.u8(CONNECTION_STATUS_TAG_CONNECTING), @@ -524,7 +524,7 @@ fn encode_connection_status(encoder: &mut Encoder, status: SocketConnectionStatu } } -fn decode_connection_status( +pub(super) fn decode_connection_status( decoder: &mut Decoder<'_>, ) -> Result { match decoder.u8()? { diff --git a/litebox_broker_protocol/src/wire/unix_socket.rs b/litebox_broker_protocol/src/wire/unix_socket.rs index c8ea3bea80..5f034eae5e 100644 --- a/litebox_broker_protocol/src/wire/unix_socket.rs +++ b/litebox_broker_protocol/src/wire/unix_socket.rs @@ -7,17 +7,22 @@ use super::fs::{ }; use super::primitive::{Decoder, Encoder}; use super::socket::{ - decode_bool, decode_option_name, decode_option_value, decode_shutdown_mode, decode_socket_type, - encode_bool, encode_option_name, encode_option_value, encode_shutdown_mode, encode_socket_type, + decode_bool, decode_connection_status, decode_option_name, decode_option_value, + decode_shutdown_mode, decode_socket_type, encode_bool, encode_connection_status, + encode_option_name, encode_option_value, encode_shutdown_mode, encode_socket_type, }; use crate::message::{UnixSocketRequest, UnixSocketResponse}; +use crate::socket::SocketConnectionStatus; use crate::unix_socket::{ AcceptUnixSocketRequest, AcceptUnixSocketResponse, BindUnixSocketRequest, - ConnectUnixSocketRequest, CreateUnixSocketPairResponse, CreateUnixSocketRequest, - CreateUnixSocketResponse, GetUnixSocketNameRequest, GetUnixSocketNameResponse, - GetUnixSocketOptionRequest, GetUnixSocketOptionResponse, ListenUnixSocketRequest, - ReceiveUnixSocketRequest, ReceiveUnixSocketResponse, SendUnixSocketRequest, - SendUnixSocketResponse, SetUnixSocketOptionRequest, ShutdownUnixSocketRequest, UnixSocketError, + CancelUnixSocketConnectRequest, CheckUnixSocketConnectRequest, CheckUnixSocketConnectResponse, + ConnectUnixSocketRequest, ConnectUnixSocketResponse, CreateUnixSocketPairResponse, + CreateUnixSocketRequest, CreateUnixSocketResponse, CreateVsockSocketRequest, + GetUnixSocketNameRequest, GetUnixSocketNameResponse, GetUnixSocketOptionRequest, + GetUnixSocketOptionResponse, ListenUnixSocketRequest, ReceiveUnixSocketRequest, + ReceiveUnixSocketResponse, SendUnixSocketRequest, SendUnixSocketResponse, + SetUnixSocketOptionRequest, ShutdownUnixSocketRequest, TakeUnixSocketErrorRequest, + TakeUnixSocketErrorResponse, UnixSocketError, }; const TAG_CREATE: u8 = 0; @@ -33,6 +38,10 @@ const TAG_GET_NAME: u8 = 9; const TAG_SET_OPTION: u8 = 10; const TAG_GET_OPTION: u8 = 11; const TAG_FAILED: u8 = 12; +const TAG_TAKE_ERROR: u8 = 13; +const TAG_CANCEL_CONNECT: u8 = 14; +const TAG_CHECK_CONNECT: u8 = 15; +const TAG_CREATE_VSOCK: u8 = 16; const ERROR_TAG_ADDRESS_IN_USE: u8 = 0; const ERROR_TAG_CONNECTION_REFUSED: u8 = 1; @@ -46,6 +55,10 @@ const ERROR_TAG_BROKEN_PIPE: u8 = 8; const ERROR_TAG_FILE: u8 = 9; const ERROR_TAG_NOT_PERMITTED: u8 = 10; const ERROR_TAG_CONNECTION_RESET: u8 = 11; +const ERROR_TAG_ADDRESS_NOT_AVAILABLE: u8 = 12; +const ERROR_TAG_UNREACHABLE: u8 = 13; +const ERROR_TAG_ALREADY_IN_PROGRESS: u8 = 14; +const ERROR_TAG_TIMED_OUT: u8 = 15; pub(super) fn encode_unix_socket_request(encoder: &mut Encoder, request: UnixSocketRequest) { match request { @@ -57,6 +70,10 @@ pub(super) fn encode_unix_socket_request(encoder: &mut Encoder, request: UnixSoc encoder.u8(TAG_CREATE_PAIR); encode_create_request(encoder, request); } + UnixSocketRequest::CreateVsock(request) => { + encoder.u8(TAG_CREATE_VSOCK); + encoder.u16(request.flags.bits()); + } UnixSocketRequest::Bind(request) => { encoder.u8(TAG_BIND); encoder.handle(request.handle); @@ -117,6 +134,18 @@ pub(super) fn encode_unix_socket_request(encoder: &mut Encoder, request: UnixSoc encoder.handle(request.handle); encode_option_name(encoder, request.name); } + UnixSocketRequest::TakeError(request) => { + encoder.u8(TAG_TAKE_ERROR); + encoder.handle(request.handle); + } + UnixSocketRequest::CheckConnect(request) => { + encoder.u8(TAG_CHECK_CONNECT); + encoder.handle(request.handle); + } + UnixSocketRequest::CancelConnect(request) => { + encoder.u8(TAG_CANCEL_CONNECT); + encoder.handle(request.handle); + } } } @@ -126,6 +155,9 @@ pub(super) fn decode_unix_socket_request( Ok(match decoder.u8()? { TAG_CREATE => UnixSocketRequest::Create(decode_create_request(decoder)?), TAG_CREATE_PAIR => UnixSocketRequest::CreatePair(decode_create_request(decoder)?), + TAG_CREATE_VSOCK => UnixSocketRequest::CreateVsock(CreateVsockSocketRequest { + flags: decode_open_flags(decoder)?, + }), TAG_BIND => UnixSocketRequest::Bind(BindUnixSocketRequest { handle: decoder.handle()?, address: decoder.shared_buffer_sequence()?, @@ -176,6 +208,15 @@ pub(super) fn decode_unix_socket_request( handle: decoder.handle()?, name: decode_option_name(decoder)?, }), + TAG_TAKE_ERROR => UnixSocketRequest::TakeError(TakeUnixSocketErrorRequest { + handle: decoder.handle()?, + }), + TAG_CHECK_CONNECT => UnixSocketRequest::CheckConnect(CheckUnixSocketConnectRequest { + handle: decoder.handle()?, + }), + TAG_CANCEL_CONNECT => UnixSocketRequest::CancelConnect(CancelUnixSocketConnectRequest { + handle: decoder.handle()?, + }), _ => return Err(WireError::InvalidTag), }) } @@ -191,9 +232,16 @@ pub(super) fn encode_unix_socket_response(encoder: &mut Encoder, response: UnixS encoder.handle(response.first); encoder.handle(response.second); } + UnixSocketResponse::CreateVsock(response) => { + encoder.u8(TAG_CREATE_VSOCK); + encoder.handle(response.handle); + } UnixSocketResponse::Bind => encoder.u8(TAG_BIND), UnixSocketResponse::Listen => encoder.u8(TAG_LISTEN), - UnixSocketResponse::Connect => encoder.u8(TAG_CONNECT), + UnixSocketResponse::Connect(response) => { + encoder.u8(TAG_CONNECT); + encode_connection_status(encoder, response.status); + } UnixSocketResponse::Accept(response) => { encoder.u8(TAG_ACCEPT); encoder.handle(response.handle); @@ -218,6 +266,18 @@ pub(super) fn encode_unix_socket_response(encoder: &mut Encoder, response: UnixS encoder.u8(TAG_GET_OPTION); encode_option_value(encoder, response.value); } + UnixSocketResponse::TakeError(response) => { + encoder.u8(TAG_TAKE_ERROR); + encode_bool(encoder, response.error.is_some()); + if let Some(error) = response.error { + encode_error(encoder, error); + } + } + UnixSocketResponse::CheckConnect(response) => { + encoder.u8(TAG_CHECK_CONNECT); + encode_connection_status(encoder, response.status); + } + UnixSocketResponse::CancelConnect => encoder.u8(TAG_CANCEL_CONNECT), UnixSocketResponse::Failed(error) => { encoder.u8(TAG_FAILED); encode_error(encoder, error); @@ -236,9 +296,21 @@ pub(super) fn decode_unix_socket_response( first: decoder.handle()?, second: decoder.handle()?, }), + TAG_CREATE_VSOCK => UnixSocketResponse::CreateVsock(CreateUnixSocketResponse { + handle: decoder.handle()?, + }), TAG_BIND => UnixSocketResponse::Bind, TAG_LISTEN => UnixSocketResponse::Listen, - TAG_CONNECT => UnixSocketResponse::Connect, + TAG_CONNECT => UnixSocketResponse::Connect(ConnectUnixSocketResponse { + // Only these complete a connect. + status: match decode_connection_status(decoder)? { + status @ (SocketConnectionStatus::Connected + | SocketConnectionStatus::Connecting) => status, + SocketConnectionStatus::Unconnected | SocketConnectionStatus::Failed(_) => { + return Err(WireError::InvalidTag); + } + }, + }), TAG_ACCEPT => UnixSocketResponse::Accept(AcceptUnixSocketResponse { handle: decoder.handle()?, }), @@ -258,6 +330,21 @@ pub(super) fn decode_unix_socket_response( TAG_GET_OPTION => UnixSocketResponse::GetOption(GetUnixSocketOptionResponse { value: decode_option_value(decoder)?, }), + TAG_TAKE_ERROR => UnixSocketResponse::TakeError(TakeUnixSocketErrorResponse { + error: decode_bool(decoder)? + .then(|| decode_error(decoder)) + .transpose()?, + }), + TAG_CHECK_CONNECT => UnixSocketResponse::CheckConnect(CheckUnixSocketConnectResponse { + // A failure is reported as one. + status: match decode_connection_status(decoder)? { + status @ (SocketConnectionStatus::Connected + | SocketConnectionStatus::Connecting + | SocketConnectionStatus::Unconnected) => status, + SocketConnectionStatus::Failed(_) => return Err(WireError::InvalidTag), + }, + }), + TAG_CANCEL_CONNECT => UnixSocketResponse::CancelConnect, TAG_FAILED => UnixSocketResponse::Failed(decode_error(decoder)?), _ => return Err(WireError::InvalidTag), }) @@ -288,6 +375,10 @@ fn encode_error(encoder: &mut Encoder, error: UnixSocketError) { UnixSocketError::BrokenPipe => encoder.u8(ERROR_TAG_BROKEN_PIPE), UnixSocketError::NotPermitted => encoder.u8(ERROR_TAG_NOT_PERMITTED), UnixSocketError::ConnectionReset => encoder.u8(ERROR_TAG_CONNECTION_RESET), + UnixSocketError::AddressNotAvailable => encoder.u8(ERROR_TAG_ADDRESS_NOT_AVAILABLE), + UnixSocketError::Unreachable => encoder.u8(ERROR_TAG_UNREACHABLE), + UnixSocketError::AlreadyInProgress => encoder.u8(ERROR_TAG_ALREADY_IN_PROGRESS), + UnixSocketError::TimedOut => encoder.u8(ERROR_TAG_TIMED_OUT), UnixSocketError::File(error) => { encoder.u8(ERROR_TAG_FILE); encode_file_error(encoder, error); @@ -308,6 +399,10 @@ fn decode_error(decoder: &mut Decoder<'_>) -> Result ERROR_TAG_BROKEN_PIPE => UnixSocketError::BrokenPipe, ERROR_TAG_NOT_PERMITTED => UnixSocketError::NotPermitted, ERROR_TAG_CONNECTION_RESET => UnixSocketError::ConnectionReset, + ERROR_TAG_ADDRESS_NOT_AVAILABLE => UnixSocketError::AddressNotAvailable, + ERROR_TAG_UNREACHABLE => UnixSocketError::Unreachable, + ERROR_TAG_ALREADY_IN_PROGRESS => UnixSocketError::AlreadyInProgress, + ERROR_TAG_TIMED_OUT => UnixSocketError::TimedOut, ERROR_TAG_FILE => UnixSocketError::File(decode_file_error(decoder)?), _ => return Err(WireError::InvalidTag), }) diff --git a/litebox_common_linux/src/errno/mod.rs b/litebox_common_linux/src/errno/mod.rs index a5e0af8026..c1a30fb28a 100644 --- a/litebox_common_linux/src/errno/mod.rs +++ b/litebox_common_linux/src/errno/mod.rs @@ -839,6 +839,10 @@ impl From for Errno { SocketError::BrokenPipe => Errno::EPIPE, SocketError::NotPermitted => Errno::EPERM, SocketError::ConnectionReset => Errno::ECONNRESET, + SocketError::AddressNotAvailable => Errno::EADDRNOTAVAIL, + SocketError::Unreachable => Errno::ENETUNREACH, + SocketError::AlreadyInProgress => Errno::EALREADY, + SocketError::TimedOut => Errno::ETIMEDOUT, SocketError::File(error) => match error { FileError::NoSuchFileOrDirectory | FileError::MissingComponent => Errno::ENOENT, FileError::AccessNotAllowed @@ -853,6 +857,7 @@ impl From for Errno { _ => Errno::EIO, }, UnixSocketError::WouldBlock => Errno::EAGAIN, + UnixSocketError::InProgress => Errno::EINPROGRESS, UnixSocketError::Interrupted => Errno::EINTR, UnixSocketError::WaitError(error) => match error { litebox::event::wait::WaitError::Interrupted => Errno::ERESTARTSYS, diff --git a/litebox_shim_linux/src/syscalls/unix.rs b/litebox_shim_linux/src/syscalls/unix.rs index 75e160a7d8..6f6573d6aa 100644 --- a/litebox_shim_linux/src/syscalls/unix.rs +++ b/litebox_shim_linux/src/syscalls/unix.rs @@ -150,7 +150,9 @@ impl UnixSocketAddr { impl From for UnixSocketAddr { fn from(name: UnixSocketName) -> Self { match name { - UnixSocketName::Unnamed => UnixSocketAddr::Unnamed, + // Only sockets bound or connected to vsock addresses, which the shim does not use, + // hold vsock names. + UnixSocketName::Unnamed | UnixSocketName::Vsock { .. } => UnixSocketAddr::Unnamed, UnixSocketName::Path(name) => { UnixSocketAddr::Path(String::from_utf8_lossy(&name).into_owned()) }