diff --git a/crates/kernel/src/pipe.rs b/crates/kernel/src/pipe.rs index ec5e733f0..8495bf006 100644 --- a/crates/kernel/src/pipe.rs +++ b/crates/kernel/src/pipe.rs @@ -60,6 +60,11 @@ pub struct PipeBuffer { len: usize, read_count: u32, write_count: u32, + /// Endpoint references owned by descriptors queued in SCM_RIGHTS messages. + /// These are included in read_count/write_count, but tracked separately so + /// unreachable cycles of queued socket descriptors can be collected. + in_flight_read_count: u32, + in_flight_write_count: u32, /// The receive half of a normally closed TCP endpoint remains as an /// orphaned discard sink until the peer closes its write half. This models /// TCP's simplex FIN without inventing a fixed number of successful writes @@ -84,6 +89,8 @@ impl PipeBuffer { len: 0, read_count: 1, write_count: 1, + in_flight_read_count: 0, + in_flight_write_count: 0, orphaned_read: false, pipe_idx: 0, ancillary_fds: VecDeque::new(), @@ -235,6 +242,48 @@ impl PipeBuffer { self.write_count += 1; } + fn retain_in_flight_reader(&mut self) { + self.add_reader(); + self.in_flight_read_count += 1; + } + + fn retain_in_flight_writer(&mut self) { + self.add_writer(); + self.in_flight_write_count += 1; + } + + fn adopt_in_flight_reader(&mut self) { + debug_assert!(self.in_flight_read_count > 0); + self.in_flight_read_count = self.in_flight_read_count.saturating_sub(1); + } + + fn adopt_in_flight_writer(&mut self) { + debug_assert!(self.in_flight_write_count > 0); + self.in_flight_write_count = self.in_flight_write_count.saturating_sub(1); + } + + fn release_in_flight_reader(&mut self) { + debug_assert!(self.in_flight_read_count > 0); + self.in_flight_read_count = self.in_flight_read_count.saturating_sub(1); + self.close_read_end(); + } + + fn release_in_flight_reader_orderly(&mut self) { + debug_assert!(self.in_flight_read_count > 0); + self.in_flight_read_count = self.in_flight_read_count.saturating_sub(1); + self.close_read_end_orderly(); + } + + fn release_in_flight_writer(&mut self) { + debug_assert!(self.in_flight_write_count > 0); + self.in_flight_write_count = self.in_flight_write_count.saturating_sub(1); + self.close_write_end(); + } + + fn has_external_reader(&self) -> bool { + self.read_count > self.in_flight_read_count + } + /// Returns true if the read end is still open (any readers remain). pub fn is_read_end_open(&self) -> bool { self.read_count > 0 || self.orphaned_read @@ -259,7 +308,7 @@ impl PipeBuffer { } /// Push ancillary FDs (SCM_RIGHTS) to be delivered with the next recvmsg. - pub fn push_ancillary(&mut self, fds: Vec) { + fn push_ancillary(&mut self, fds: Vec) { if !fds.is_empty() { self.ancillary_fds.push_back(fds); } @@ -351,16 +400,269 @@ impl PipeTable { self.pipes.get_mut(idx).and_then(|p| p.as_mut()) } + /// Queue one SCM_RIGHTS message and retain every pipe-backed resource it + /// carries. The message is collected immediately if no live or reachable + /// socket endpoint can ever receive it. + pub fn queue_ancillary(&mut self, carrier_idx: usize, fds: Vec) -> bool { + if self.get(carrier_idx).is_none() || !self.retain_ancillary_resources(&fds) { + return false; + } + self.get_mut(carrier_idx).unwrap().push_ancillary(fds); + self.collect_unreachable_ancillary(); + true + } + + /// Retain every pipe-backed resource carried by one SCM_RIGHTS message. + /// + /// A successful send owns these references while the descriptors are in + /// transit. Receiving a descriptor transfers the matching reference to + /// the new OFD; discarding the message releases it instead. + pub fn retain_ancillary_resources(&mut self, fds: &[InFlightFd]) -> bool { + for (retained, fd) in fds.iter().enumerate() { + if !self.retain_ancillary_resource(fd) { + for retained_fd in &fds[..retained] { + self.release_ancillary_resource_inner(retained_fd); + } + self.collect_unreachable_ancillary(); + return false; + } + } + true + } + + /// Release pipe-backed resources for SCM_RIGHTS descriptors that were not + /// installed in a receiving process. + pub fn release_ancillary_resources(&mut self, fds: &[InFlightFd]) { + for fd in fds { + self.release_ancillary_resource_inner(fd); + } + self.collect_unreachable_ancillary(); + } + + /// Transfer one retained SCM_RIGHTS reference into a newly installed OFD. + /// Call finish_ancillary_transition after every entry in the popped batch + /// has either been adopted or released. + pub fn adopt_ancillary_resource(&mut self, fd: &InFlightFd) { + self.adopt_ancillary_resource_inner(fd); + } + + /// Release one entry from a popped SCM_RIGHTS batch that could not be + /// installed. Collection is deferred until the entire batch is resolved. + pub fn release_ancillary_resource(&mut self, fd: &InFlightFd) { + self.release_ancillary_resource_inner(fd); + } + + /// Complete a popped SCM_RIGHTS batch after every reference was adopted or + /// released, then collect cycles made unreachable by removing that queue. + pub fn finish_ancillary_transition(&mut self) { + self.collect_unreachable_ancillary(); + } + + fn retain_ancillary_resource(&mut self, fd: &InFlightFd) -> bool { + if fd.file_type == 0 && fd.host_handle < 0 { + let pipe_idx = (-(fd.host_handle + 1)) as usize; + let Some(pipe) = self.get_mut(pipe_idx) else { + return false; + }; + match fd.status_flags & wasm_posix_shared::flags::O_ACCMODE { + wasm_posix_shared::flags::O_RDONLY => pipe.retain_in_flight_reader(), + wasm_posix_shared::flags::O_WRONLY => pipe.retain_in_flight_writer(), + wasm_posix_shared::flags::O_RDWR => { + pipe.retain_in_flight_reader(); + pipe.retain_in_flight_writer(); + } + _ => return false, + } + } else if fd.file_type == 1 { + let Some(socket) = fd.socket.as_ref() else { + return true; + }; + if !socket.global_pipes { + return true; + } + + if let Some(send_idx) = socket.send_buf_idx { + let Some(pipe) = self.get_mut(send_idx) else { + return false; + }; + pipe.retain_in_flight_writer(); + } + if let Some(recv_idx) = socket.recv_buf_idx { + let Some(pipe) = self.get_mut(recv_idx) else { + if let Some(send_idx) = socket.send_buf_idx { + if let Some(pipe) = self.get_mut(send_idx) { + pipe.release_in_flight_writer(); + } + self.free_fully_closed_inner(send_idx); + } + return false; + }; + pipe.retain_in_flight_reader(); + } + } + true + } + + fn adopt_ancillary_resource_inner(&mut self, fd: &InFlightFd) { + if fd.file_type == 0 && fd.host_handle < 0 { + let pipe_idx = (-(fd.host_handle + 1)) as usize; + if let Some(pipe) = self.get_mut(pipe_idx) { + match fd.status_flags & wasm_posix_shared::flags::O_ACCMODE { + wasm_posix_shared::flags::O_RDONLY => pipe.adopt_in_flight_reader(), + wasm_posix_shared::flags::O_WRONLY => pipe.adopt_in_flight_writer(), + wasm_posix_shared::flags::O_RDWR => { + pipe.adopt_in_flight_reader(); + pipe.adopt_in_flight_writer(); + } + _ => {} + } + } + } else if fd.file_type == 1 { + let Some(socket) = fd.socket.as_ref() else { + return; + }; + if !socket.global_pipes { + return; + } + if let Some(send_idx) = socket.send_buf_idx { + if let Some(pipe) = self.get_mut(send_idx) { + pipe.adopt_in_flight_writer(); + } + } + if let Some(recv_idx) = socket.recv_buf_idx { + if let Some(pipe) = self.get_mut(recv_idx) { + pipe.adopt_in_flight_reader(); + } + } + } + } + + fn release_ancillary_resource_inner(&mut self, fd: &InFlightFd) { + if fd.file_type == 0 && fd.host_handle < 0 { + let pipe_idx = (-(fd.host_handle + 1)) as usize; + if let Some(pipe) = self.get_mut(pipe_idx) { + match fd.status_flags & wasm_posix_shared::flags::O_ACCMODE { + wasm_posix_shared::flags::O_RDONLY => pipe.release_in_flight_reader(), + wasm_posix_shared::flags::O_WRONLY => pipe.release_in_flight_writer(), + wasm_posix_shared::flags::O_RDWR => { + pipe.release_in_flight_reader(); + pipe.release_in_flight_writer(); + } + _ => {} + } + } + self.free_fully_closed_inner(pipe_idx); + } else if fd.file_type == 1 { + let Some(socket) = fd.socket.as_ref() else { + return; + }; + if !socket.global_pipes { + return; + } + + if let Some(send_idx) = socket.send_buf_idx { + if let Some(pipe) = self.get_mut(send_idx) { + pipe.release_in_flight_writer(); + } + self.free_fully_closed_inner(send_idx); + } + if let Some(recv_idx) = socket.recv_buf_idx { + if let Some(pipe) = self.get_mut(recv_idx) { + let orderly_tcp_close = socket.sock_type == 0 + && matches!(socket.domain, 1 | 2); + if orderly_tcp_close { + pipe.release_in_flight_reader_orderly(); + } else { + pipe.release_in_flight_reader(); + } + } + self.free_fully_closed_inner(recv_idx); + } + } + } + /// Free a pipe buffer slot if both endpoints are closed. pub fn free_if_closed(&mut self, idx: usize) { - if let Some(Some(pipe)) = self.pipes.get(idx) { - if pipe.is_fully_closed() { - self.pipes[idx] = None; - self.free_list.push(idx); + self.free_fully_closed_inner(idx); + self.collect_unreachable_ancillary(); + } + + fn free_fully_closed_inner(&mut self, idx: usize) { + let should_free = self + .pipes + .get(idx) + .and_then(Option::as_ref) + .is_some_and(PipeBuffer::is_fully_closed); + if !should_free { + return; + } + + let pipe = self.pipes[idx].take().unwrap(); + self.free_list.push(idx); + for fds in pipe.ancillary_fds { + for fd in &fds { + self.release_ancillary_resource_inner(fd); } } } + /// Collect ancillary queues that cannot be reached from an externally + /// owned receive endpoint. Queued socket descriptors form graph edges to + /// their receive pipes; the mark phase preserves every transitively + /// receivable cycle and the sweep drops only components with no root. + fn collect_unreachable_ancillary(&mut self) { + let mut reachable = Vec::new(); + reachable.resize(self.pipes.len(), false); + let mut work = VecDeque::new(); + for (idx, pipe) in self.pipes.iter().enumerate() { + if pipe.as_ref().is_some_and(PipeBuffer::has_external_reader) { + reachable[idx] = true; + work.push_back(idx); + } + } + + while let Some(carrier_idx) = work.pop_front() { + let Some(pipe) = self.get(carrier_idx) else { + continue; + }; + for batch in &pipe.ancillary_fds { + for fd in batch { + if fd.file_type != 1 { + continue; + } + let Some(socket) = fd.socket.as_ref() else { + continue; + }; + if !socket.global_pipes { + continue; + } + let Some(recv_idx) = socket.recv_buf_idx else { + continue; + }; + if recv_idx < reachable.len() && !reachable[recv_idx] { + reachable[recv_idx] = true; + work.push_back(recv_idx); + } + } + } + } + + let mut dropped = Vec::new(); + for (idx, pipe) in self.pipes.iter_mut().enumerate() { + if !reachable[idx] { + if let Some(pipe) = pipe.as_mut() { + dropped.extend(pipe.ancillary_fds.drain(..)); + } + } + } + for batch in dropped { + for fd in &batch { + self.release_ancillary_resource_inner(fd); + } + } + + } + /// Release both endpoints of a newly allocated buffer that was never /// published to a socket or host bridge, then make its slot reusable. pub fn discard_unclaimed(&mut self, idx: usize) { @@ -619,4 +921,134 @@ mod tests { assert_eq!(table.count_active(), 0); assert_eq!(table.alloc(PipeBuffer::new(64)), idx); } + + fn in_flight_pipe_read_end(pipe_idx: usize) -> InFlightFd { + InFlightFd { + file_type: 0, + status_flags: wasm_posix_shared::flags::O_RDONLY, + host_handle: -((pipe_idx as i64) + 1), + offset: 0, + path: b"/dev/pipe".to_vec(), + socket: None, + } + } + + fn in_flight_socket(send_idx: usize, recv_idx: usize) -> InFlightFd { + InFlightFd { + file_type: 1, + status_flags: wasm_posix_shared::flags::O_RDWR, + host_handle: -1, + offset: 0, + path: b"socket".to_vec(), + socket: Some(InFlightSocket { + domain: 0, + sock_type: 0, + protocol: 0, + state: 3, + send_buf_idx: Some(send_idx), + recv_buf_idx: Some(recv_idx), + global_pipes: true, + shut_rd: false, + shut_wr: false, + bind_addr: [0; 4], + bind_port: 0, + peer_addr: [0; 4], + peer_port: 0, + }), + } + } + + fn close_external_endpoints(table: &mut PipeTable, indices: &[usize]) { + for &idx in indices { + if let Some(pipe) = table.get_mut(idx) { + pipe.close_read_end(); + pipe.close_write_end(); + } + table.free_if_closed(idx); + } + } + + #[test] + fn scm_rights_reference_becomes_received_pipe_endpoint() { + let mut table = PipeTable::new(); + let pipe_idx = table.alloc(PipeBuffer::new(64)); + let right = in_flight_pipe_read_end(pipe_idx); + + assert!(table.retain_ancillary_resources(core::slice::from_ref(&right))); + table.adopt_ancillary_resource(&right); + table.finish_ancillary_transition(); + table.get_mut(pipe_idx).unwrap().close_read_end(); + assert_eq!(table.get_mut(pipe_idx).unwrap().write(b"still connected"), 15); + + // Receiving transfers the retained reference to the new OFD. Its final + // close, not installation, consumes that same reference. + table.get_mut(pipe_idx).unwrap().close_write_end(); + let mut payload = [0u8; 15]; + assert_eq!(table.get_mut(pipe_idx).unwrap().read(&mut payload), 15); + assert_eq!(&payload, b"still connected"); + table.get_mut(pipe_idx).unwrap().close_read_end(); + table.free_if_closed(pipe_idx); + assert!(table.get(pipe_idx).is_none()); + } + + #[test] + fn closing_carrier_releases_queued_pipe_endpoint() { + let mut table = PipeTable::new(); + let pipe_idx = table.alloc(PipeBuffer::new(64)); + let carrier_idx = table.alloc(PipeBuffer::new(64)); + let right = in_flight_pipe_read_end(pipe_idx); + + assert!(table.queue_ancillary(carrier_idx, vec![right])); + table.get_mut(pipe_idx).unwrap().close_read_end(); + assert_eq!(table.get_mut(pipe_idx).unwrap().write(b"held"), 4); + + table.get_mut(carrier_idx).unwrap().close_read_end(); + table.get_mut(carrier_idx).unwrap().close_write_end(); + table.free_if_closed(carrier_idx); + assert_eq!(table.get_mut(pipe_idx).unwrap().write(b"released"), 0); + + table.get_mut(pipe_idx).unwrap().close_write_end(); + table.free_if_closed(pipe_idx); + assert!(table.get(pipe_idx).is_none()); + } + + #[test] + fn unreachable_self_socket_right_cycle_is_collected_and_slots_reused() { + let mut table = PipeTable::new(); + + for _ in 0..32 { + let (carrier_idx, peer_send_idx) = + table.alloc_pair(PipeBuffer::new(64), PipeBuffer::new(64)); + assert_eq!((carrier_idx, peer_send_idx), (0, 1)); + + // The peer socket receives from carrier_idx. Queuing that peer on + // carrier_idx creates the canonical SCM_RIGHTS self-cycle. + let peer = in_flight_socket(peer_send_idx, carrier_idx); + assert!(table.queue_ancillary(carrier_idx, vec![peer])); + assert!(table.get(carrier_idx).unwrap().has_ancillary()); + + close_external_endpoints(&mut table, &[carrier_idx, peer_send_idx]); + assert_eq!(table.count_active(), 0); + } + } + + #[test] + fn unreachable_cross_socket_right_cycle_is_collected() { + let mut table = PipeTable::new(); + let (a_recv_idx, a_send_idx) = + table.alloc_pair(PipeBuffer::new(64), PipeBuffer::new(64)); + let (b_recv_idx, b_send_idx) = + table.alloc_pair(PipeBuffer::new(64), PipeBuffer::new(64)); + + let a = in_flight_socket(a_send_idx, a_recv_idx); + let b = in_flight_socket(b_send_idx, b_recv_idx); + assert!(table.queue_ancillary(a_recv_idx, vec![b])); + assert!(table.queue_ancillary(b_recv_idx, vec![a])); + + close_external_endpoints( + &mut table, + &[a_recv_idx, a_send_idx, b_recv_idx, b_send_idx], + ); + assert_eq!(table.count_active(), 0); + } } diff --git a/crates/kernel/src/wasm_api.rs b/crates/kernel/src/wasm_api.rs index c2f5ddfe8..41d1d9e13 100644 --- a/crates/kernel/src/wasm_api.rs +++ b/crates/kernel/src/wasm_api.rs @@ -6575,6 +6575,37 @@ fn extract_scm_rights( result } +/// Queue one successfully sent SCM_RIGHTS message and retain the resources +/// represented by its serialized descriptors until receive or discard. +fn queue_scm_rights_fds( + proc: &crate::process::Process, + socket_fd: i32, + fds: Vec, +) -> bool { + let send_idx = { + let Ok(fd_entry) = proc.fd_table.get(socket_fd) else { + return false; + }; + let Some(ofd) = proc.ofd_table.get(fd_entry.ofd_ref.0) else { + return false; + }; + if ofd.file_type != crate::ofd::FileType::Socket { + return false; + } + let sock_idx = (-(ofd.host_handle + 1)) as usize; + let Some(socket) = proc.sockets.get(sock_idx) else { + return false; + }; + let Some(send_idx) = socket.send_buf_idx else { + return false; + }; + send_idx + }; + + let pipes = unsafe { crate::pipe::global_pipe_table() }; + pipes.queue_ancillary(send_idx, fds) +} + /// sendmsg — send a message on a socket. /// Parses msghdr to extract iov[0] and delegates to sys_sendmsg. /// Handles SCM_RIGHTS ancillary data by serializing FDs into the pipe's ancillary queue. @@ -6622,45 +6653,10 @@ pub extern "C" fn kernel_sendmsg(fd: i32, msg_ptr: *const u8, flags: u32) -> i32 } }; - // If data was sent successfully and we have ancillary FDs, push them to the pipe + // If data was sent successfully, the queued SCM_RIGHTS message owns one + // resource reference per descriptor until recvmsg installs or discards it. if result > 0 && !ancillary_fds.is_empty() { - // Increment pipe refcounts for socket FDs being passed BEFORE - // borrowing the send pipe (avoids double &mut borrow of global_pipe_table). - for in_flight in &ancillary_fds { - if let Some(ref sock_info) = in_flight.socket { - if sock_info.global_pipes { - let pt = unsafe { crate::pipe::global_pipe_table() }; - if let Some(idx) = sock_info.send_buf_idx { - if let Some(p) = pt.get_mut(idx) { - p.add_writer(); - } - } - if let Some(idx) = sock_info.recv_buf_idx { - if let Some(p) = pt.get_mut(idx) { - p.add_reader(); - } - } - } - } - } - - // Now push ancillary data to the socket's send pipe - if let Ok(fd_entry) = proc.fd_table.get(fd) { - if let Some(ofd) = proc.ofd_table.get(fd_entry.ofd_ref.0) { - if ofd.file_type == crate::ofd::FileType::Socket { - let sock_idx = (-(ofd.host_handle + 1)) as usize; - if let Some(sock) = proc.sockets.get(sock_idx) { - if let Some(send_idx) = sock.send_buf_idx { - let pipe = - unsafe { crate::pipe::global_pipe_table().get_mut(send_idx) }; - if let Some(pipe) = pipe { - pipe.push_ancillary(ancillary_fds); - } - } - } - } - } - } + let _ = queue_scm_rights_fds(proc, fd, ancillary_fds); } deliver_pending_signals(proc, &mut host); @@ -6721,22 +6717,53 @@ fn install_scm_rights_fds( new_host_handle, entry.path.clone(), ); - if let Ok(new_fd) = proc.fd_table.alloc(crate::fd::OpenFileDescRef(ofd_idx), 0) + match proc + .fd_table + .alloc(crate::fd::OpenFileDescRef(ofd_idx), 0) { - new_fds.push(new_fd); + Ok(new_fd) => { + unsafe { + crate::pipe::global_pipe_table().adopt_ancillary_resource(&entry); + } + new_fds.push(new_fd); + } + Err(_) => { + proc.ofd_table.dec_ref(ofd_idx); + proc.sockets.free(new_sock_idx); + unsafe { + crate::pipe::global_pipe_table() + .release_ancillary_resource(&entry); + } + } } } } 0 => { - // Pipe FD + // The reference retained by sendmsg becomes this OFD's + // endpoint reference. Do not increment it a second time. let ofd_idx = proc.ofd_table.create( FileType::Pipe, entry.status_flags, entry.host_handle, entry.path.clone(), ); - if let Ok(new_fd) = proc.fd_table.alloc(crate::fd::OpenFileDescRef(ofd_idx), 0) { - new_fds.push(new_fd); + match proc + .fd_table + .alloc(crate::fd::OpenFileDescRef(ofd_idx), 0) + { + Ok(new_fd) => { + unsafe { + crate::pipe::global_pipe_table().adopt_ancillary_resource(&entry); + } + new_fds.push(new_fd); + } + Err(_) => { + proc.ofd_table.dec_ref(ofd_idx); + unsafe { + crate::pipe::global_pipe_table() + .release_ancillary_resource(&entry); + } + } } } 6 | 7 => { @@ -6789,6 +6816,9 @@ fn install_scm_rights_fds( } } + unsafe { + crate::pipe::global_pipe_table().finish_ancillary_transition(); + } new_fds } @@ -6849,6 +6879,7 @@ pub extern "C" fn kernel_recvmsg(fd: i32, msg_ptr: *mut u8, flags: u32) -> i32 { // Pop ancillary data in a limited scope to avoid holding &mut pipe across // install_scm_rights_fds (which modifies proc). let mut ancillary_delivered = false; + let mut output_msg_flags = 0u32; if result > 0 { let popped = 'pop: { let recv_idx = { @@ -6877,32 +6908,62 @@ pub extern "C" fn kernel_recvmsg(fd: i32, msg_ptr: *mut u8, flags: u32) -> i32 { } }; - if let Some(in_flight) = popped { - let new_fds = install_scm_rights_fds(proc, in_flight); - // Build cmsg response in msg_control buffer - if control_ptr != 0 && control_len > 0 && !new_fds.is_empty() { + if let Some(mut in_flight) = popped { + let control_fd_capacity = if control_ptr != 0 && control_len >= 16 { + (control_len - 12) / 4 + } else { + 0 + }; + let install_count = control_fd_capacity.min(in_flight.len()); + let excess = in_flight.split_off(install_count); + + // Keep collection deferred while the fitting prefix is not yet + // represented by receiver OFDs. Otherwise releasing an excess + // socket could collect queues that the prefix makes reachable. + if !excess.is_empty() { + let pipes = unsafe { crate::pipe::global_pipe_table() }; + for entry in &excess { + pipes.release_ancillary_resource(entry); + } + } + + let attempted = in_flight.len(); + let new_fds = if attempted == 0 { + unsafe { + crate::pipe::global_pipe_table().finish_ancillary_transition(); + } + Vec::new() + } else { + install_scm_rights_fds(proc, in_flight) + }; + + if !excess.is_empty() || new_fds.len() < attempted { + output_msg_flags |= wasm_posix_shared::socket::MSG_CTRUNC; + } + + // Build cmsg response in msg_control buffer. + if !new_fds.is_empty() { let cmsg_data_len = new_fds.len() * 4; let cmsg_len = 12 + cmsg_data_len; // cmsghdr + FD data let cmsg_space = (cmsg_len + 3) & !3; // aligned - if cmsg_space <= control_len { - let ctrl = - unsafe { slice::from_raw_parts_mut(control_ptr as *mut u8, control_len) }; - // cmsg_len - ctrl[0..4].copy_from_slice(&(cmsg_len as u32).to_le_bytes()); - // cmsg_level = SOL_SOCKET - ctrl[4..8].copy_from_slice(&1u32.to_le_bytes()); - // cmsg_type = SCM_RIGHTS - ctrl[8..12].copy_from_slice(&1u32.to_le_bytes()); - // FD numbers - for (i, &new_fd) in new_fds.iter().enumerate() { - let off = 12 + i * 4; - ctrl[off..off + 4].copy_from_slice(&new_fd.to_le_bytes()); - } - // Set msg_controllen - let msg_mut = unsafe { slice::from_raw_parts_mut(msg_ptr, 28) }; - msg_mut[20..24].copy_from_slice(&(cmsg_space as u32).to_le_bytes()); - ancillary_delivered = true; + let ctrl = unsafe { + slice::from_raw_parts_mut(control_ptr as *mut u8, control_len) + }; + // cmsg_len + ctrl[0..4].copy_from_slice(&(cmsg_len as u32).to_le_bytes()); + // cmsg_level = SOL_SOCKET + ctrl[4..8].copy_from_slice(&1u32.to_le_bytes()); + // cmsg_type = SCM_RIGHTS + ctrl[8..12].copy_from_slice(&1u32.to_le_bytes()); + // FD numbers + for (i, &new_fd) in new_fds.iter().enumerate() { + let off = 12 + i * 4; + ctrl[off..off + 4].copy_from_slice(&new_fd.to_le_bytes()); } + // Set msg_controllen + let msg_mut = unsafe { slice::from_raw_parts_mut(msg_ptr, 28) }; + msg_mut[20..24].copy_from_slice(&(cmsg_space as u32).to_le_bytes()); + ancillary_delivered = true; } } } @@ -6912,6 +6973,8 @@ pub extern "C" fn kernel_recvmsg(fd: i32, msg_ptr: *mut u8, flags: u32) -> i32 { let msg_mut = unsafe { slice::from_raw_parts_mut(msg_ptr, 28) }; msg_mut[20..24].copy_from_slice(&0u32.to_le_bytes()); } + let msg_mut = unsafe { slice::from_raw_parts_mut(msg_ptr, 28) }; + msg_mut[24..28].copy_from_slice(&output_msg_flags.to_le_bytes()); deliver_pending_signals(proc, &mut host); result diff --git a/crates/shared/src/lib.rs b/crates/shared/src/lib.rs index dadca0dde..60b02a45c 100644 --- a/crates/shared/src/lib.rs +++ b/crates/shared/src/lib.rs @@ -750,6 +750,7 @@ pub mod socket { pub const IPV6_TCLASS: u32 = 67; pub const MSG_OOB: u32 = 1; pub const MSG_PEEK: u32 = 2; + pub const MSG_CTRUNC: u32 = 0x08; pub const MSG_TRUNC: u32 = 0x20; pub const MSG_DONTWAIT: u32 = 64; pub const MSG_NOSIGNAL: u32 = 0x4000; diff --git a/docs/posix-status.md b/docs/posix-status.md index 1f31926b6..dae1c6d93 100644 --- a/docs/posix-status.md +++ b/docs/posix-status.md @@ -250,7 +250,7 @@ shortcuts. | `connect()` | Partial | AF_UNIX streams support same- and cross-process pathname or abstract-namespace listeners; pathname lookup uses the same canonical component walker as bind, including cross-process retries. AF_UNIX datagrams deliver to a registered peer only within the same process; a missing, wrong-type, or cross-process peer returns ECONNREFUSED until machine-wide datagram routing exists. AF_INET TCP is host-backed and works over Node external TCP or the browser local virtual-network backend. For an external non-blocking TCP handshake, the first pending call reports EINPROGRESS, a repeat while it remains pending reports EALREADY, and poll reports writable when completion or failure can be collected through SO_ERROR; blocking callers wait through the same host connection. AF_INET UDP connect stores the peer, auto-binds an ephemeral local port when needed, filters receives to the connected peer, and supports AF_UNSPEC unconnect. AF_INET6 streams support same- and cross-process `::1`; AF_INET6 datagrams are process-local and report `IPV6_V6ONLY=1` because dual-stack datagram routing is not implemented. Non-loopback IPv6 fails with EADDRNOTAVAIL for streams and ENETUNREACH for datagrams. External raw UDP also returns ENETUNREACH without another HostIO transport. | | `send()` / `recv()` | Partial | Unix domain streams and datagrams, AF_INET/AF_INET6 TCP streams, and connected AF_INET/AF_INET6 UDP preserve their socket-family addressing and datagram boundaries. TCP send/recv works over Node external TCP and the local virtual-network backend. Datagram MSG_PEEK and MSG_DONTWAIT are handled through recvfrom. Normal TCP close drains queued bytes before FIN and EOF; no transport invents a fixed post-FIN write count. A send rejected by a closed/reset stream returns EPIPE and raises SIGPIPE, while direct host/virtual handles may preserve ECONNRESET; accepted pipe-bridged resets currently surface as EOF/EPIPE. MSG_NOSIGNAL suppresses SIGPIPE without changing the errno. | | `sendto()` / `recvfrom()` | Partial | AF_INET, AF_INET6, and AF_UNIX datagrams support connected and unconnected send, receive queues, and connected-peer filtering. IPv4/IPv6 return sender addresses; AF_UNIX currently returns only the family. IPv4 limited-broadcast sends to `255.255.255.255` require `SO_BROADCAST` and fail with `EACCES` without it; enabling the option passes that permission gate, after which the send reaches the active routing/backend boundary. Kandelo does not itself model broadcast delivery. On AF_INET, AF_INET6, and AF_UNIX datagrams, Linux's input `MSG_TRUNC` extension returns the full datagram length while copying at most the caller's buffer; ordinary consume/`MSG_PEEK` behavior is unchanged. IPv4/IPv6 UDP receive queues hold 128 datagrams and drop a new arrival once full, preserving the accepted queue's order; `SO_RCVBUF` requests do not size that fixed queue, and `getsockopt` reports the fixed default capacity. AF_UNIX uses the same bound but preserves reliable delivery: a full queue blocks a blocking send through host retry and returns EAGAIN for `O_NONBLOCK`/`MSG_DONTWAIT`; capacity, association, shutdown, close, and pathname changes wake blocked sends and writable readiness waits to observe capacity or the new immediate error. In-kernel IPv4/IPv6 loopback, AF_UNIX datagram, and IPv4 multicast delivery currently reaches sockets in the sender's process only; machine-wide cross-process datagram routing remains unimplemented. Fork preserves kernel-local bind reservations and lookup ownership, but it does not yet share or transfer a host-backed UDP registration. The `10.88.*` LocalVirtualNetwork path can route IPv4 datagrams between attached Kandelo machines through HostIO for the process that registered the endpoint. IPv4 multicast supports interface selection, loop suppression, membership, and source filtering only; IPv6 multicast and external raw UDP are not implemented. | -| `sendmsg()` / `recvmsg()` | Partial | Minimal first-iovec wrappers are implemented. Input `MSG_TRUNC` reaches the datagram receive behavior above, but `recvmsg()` does not yet populate output `msg_flags`, including `MSG_TRUNC`. | +| `sendmsg()` / `recvmsg()` | Partial | Minimal first-iovec wrappers are implemented. `SCM_RIGHTS` on AF_UNIX streams preserves kernel pipe and socket endpoints while descriptors are queued, transfers each reference to a successfully installed fd, releases descriptors that cannot be returned, and collects unreachable self-referential or mutually referential socket-rights cycles. Closing the sender's fd cannot invalidate an in-flight or received endpoint. `recvmsg()` installs the descriptor prefix that fits the caller's control buffer, releases the excess, and reports `MSG_CTRUNC`. Input `MSG_TRUNC` reaches the datagram receive behavior above, but `recvmsg()` does not yet populate that output flag. | | `setsockopt()` / `getsockopt()` | Partial | SOL_SOCKET exposes SO_TYPE, SO_DOMAIN, SO_ERROR, SO_ACCEPTCONN, SO_RCVBUF, and SO_SNDBUF; SO_REUSEADDR affects UDP bind conflicts. `SO_RCVTIMEO`/`SO_SNDTIMEO` accept musl's wasm32 time64 option numbers (66/67) and wasm64 long64 numbers (20/21), canonicalizing both to the same stored timeout state; `struct timeval` is 16 bytes on both ABIs. `SO_RCVBUF`/`SO_SNDBUF` requests are accepted and stored but do not resize kernel queues or pipe buffers; `getsockopt()` reports the fixed default. `SO_BROADCAST` controls only the IPv4 limited-broadcast permission gate and does not provide broadcast delivery. SO_LINGER uses `struct linger`; its disabled form is stored, while enabling timed or reset-style linger returns EOPNOTSUPP until every transport supports the close mode. SO_BINDTODEVICE validates `lo`/`eth0`, supports empty-name unbind, and constrains bind/connect/send routing. TCP_CONGESTION uses a string layout and accepts only the modeled `cubic` policy; selecting unimplemented algorithms fails. IPv4 multicast membership/source-filter options drive process-local loopback delivery. IPV6_V6ONLY controls pre-bind stream dual-stack behavior; AF_INET6 datagrams truthfully remain V6-only. Other accepted IPv6 multicast options are stored but do not provide IPv6 multicast transport. | | `shutdown()` | Partial | SHUT_RD, SHUT_WR, and SHUT_RDWR transitions are idempotent within a process and release each owned pipe/host reference once. UDP write shutdown returns EPIPE on datagram send; read shutdown is EOF-like for recv/poll. Sending to a read-shut AF_UNIX datagram peer returns EPIPE (and SIGPIPE unless MSG_NOSIGNAL is used), and the transition wakes blocked sends/readiness waits. Fork-inherited sockets still clone shutdown flags per process instead of sharing one socket-wide shutdown state, and the external host ABI has no half-shutdown operation. | | `select()` | Partial | Wrapper around poll(). Converts fd_set bitmasks to pollfd array. Timeout supported via a host retry loop. A caught signal interrupts a would-block retry, including the no-fd sleep path, with EINTR; ignored signals leave it parked and a concurrently ready result is preserved. | diff --git a/host/test/scm-rights-pipe-lifetime.test.ts b/host/test/scm-rights-pipe-lifetime.test.ts new file mode 100644 index 000000000..ab442a36d --- /dev/null +++ b/host/test/scm-rights-pipe-lifetime.test.ts @@ -0,0 +1,21 @@ +import { describe, expect, it } from "vitest"; +import { tryResolveBinary } from "../src/binary-resolver"; +import { runCentralizedProgram } from "./centralized-test-helper"; + +const program = tryResolveBinary("programs/scm-rights-pipe-lifetime.wasm"); + +describe("SCM_RIGHTS pipe endpoint lifetime", () => { + it.skipIf(!program)("transfers live endpoints and collects unreachable rights cycles", async () => { + const result = await runCentralizedProgram({ + programPath: program!, + argv: ["scm-rights-pipe-lifetime"], + timeout: 10_000, + useDefaultRootfs: false, + }); + + expect(result.exitCode, `stderr=${result.stderr}\nstdout=${result.stdout}`).toBe(0); + expect(result.stdout).toContain( + "PASS: SCM_RIGHTS owns pipe endpoints in flight and after receipt", + ); + }); +}); diff --git a/programs/scm-rights-pipe-lifetime.c b/programs/scm-rights-pipe-lifetime.c new file mode 100644 index 000000000..57c4e9af1 --- /dev/null +++ b/programs/scm-rights-pipe-lifetime.c @@ -0,0 +1,354 @@ +#include +#include +#include +#include +#include +#include +#include + +static int send_fds(int socket_fd, const int *fds, size_t count) { + if (count == 0 || count > 2) { + errno = EINVAL; + return -1; + } + char byte = 'R'; + struct iovec iov = { + .iov_base = &byte, + .iov_len = sizeof(byte), + }; + char control[CMSG_SPACE(2 * sizeof(int))]; + memset(control, 0, sizeof(control)); + struct msghdr message = { + .msg_iov = &iov, + .msg_iovlen = 1, + .msg_control = control, + .msg_controllen = CMSG_SPACE(count * sizeof(int)), + }; + struct cmsghdr *cmsg = CMSG_FIRSTHDR(&message); + cmsg->cmsg_len = CMSG_LEN(count * sizeof(int)); + cmsg->cmsg_level = SOL_SOCKET; + cmsg->cmsg_type = SCM_RIGHTS; + memcpy(CMSG_DATA(cmsg), fds, count * sizeof(int)); + return sendmsg(socket_fd, &message, 0) == 1 ? 0 : -1; +} + +static int send_fd(int socket_fd, int fd) { + return send_fds(socket_fd, &fd, 1); +} + +static int receive_fds_with_flags(int socket_fd, int *fds, size_t capacity, + int *message_flags) { + if (capacity == 0 || capacity > 2) { + errno = EINVAL; + return -1; + } + char byte = 0; + struct iovec iov = { + .iov_base = &byte, + .iov_len = sizeof(byte), + }; + char control[CMSG_SPACE(2 * sizeof(int))]; + memset(control, 0, sizeof(control)); + struct msghdr message = { + .msg_iov = &iov, + .msg_iovlen = 1, + .msg_control = control, + .msg_controllen = CMSG_SPACE(capacity * sizeof(int)), + }; + if (recvmsg(socket_fd, &message, 0) != 1) + return -1; + if (message_flags) + *message_flags = message.msg_flags; + struct cmsghdr *cmsg = CMSG_FIRSTHDR(&message); + if (!cmsg || cmsg->cmsg_level != SOL_SOCKET || + cmsg->cmsg_type != SCM_RIGHTS || + cmsg->cmsg_len < CMSG_LEN(sizeof(int))) { + errno = EBADMSG; + return -1; + } + size_t count = (cmsg->cmsg_len - CMSG_LEN(0)) / sizeof(int); + if (count == 0 || count > capacity) { + errno = EBADMSG; + return -1; + } + memcpy(fds, CMSG_DATA(cmsg), count * sizeof(int)); + return (int)count; +} + +static int receive_fds(int socket_fd, int *fds, size_t capacity) { + return receive_fds_with_flags(socket_fd, fds, capacity, NULL); +} + +static int receive_fd(int socket_fd) { + int fd = -1; + return receive_fds(socket_fd, &fd, 1) == 1 ? fd : -1; +} + +static int transferred_endpoint_survives_sender_close(void) { + int carrier[2]; + int data_pipe[2]; + if (socketpair(AF_UNIX, SOCK_STREAM, 0, carrier) < 0 || pipe(data_pipe) < 0) + return -1; + if (write(data_pipe[1], "before", 6) != 6 || + send_fd(carrier[0], data_pipe[0]) < 0 || close(data_pipe[0]) < 0 || + write(data_pipe[1], "after", 5) != 5) { + return -1; + } + + int received = receive_fd(carrier[1]); + if (received < 0 || close(data_pipe[1]) < 0) + return -1; + + char payload[11]; + ssize_t total = 0; + while (total < (ssize_t)sizeof(payload)) { + ssize_t count = read(received, payload + total, sizeof(payload) - total); + if (count <= 0) + return -1; + total += count; + } + char extra; + if (memcmp(payload, "beforeafter", sizeof(payload)) != 0 || + read(received, &extra, sizeof(extra)) != 0) { + errno = EIO; + return -1; + } + + return close(received) | close(carrier[0]) | close(carrier[1]); +} + +static int transferred_writer_survives_sender_close(void) { + int carrier[2]; + int data_pipe[2]; + if (socketpair(AF_UNIX, SOCK_STREAM, 0, carrier) < 0 || pipe(data_pipe) < 0) + return -1; + if (send_fd(carrier[0], data_pipe[1]) < 0 || close(data_pipe[1]) < 0) + return -1; + + int received = receive_fd(carrier[1]); + if (received < 0 || write(received, "writer", 6) != 6 || close(received) < 0) + return -1; + + char payload[6]; + ssize_t count = read(data_pipe[0], payload, sizeof(payload)); + char extra; + if (count != (ssize_t)sizeof(payload) || + memcmp(payload, "writer", sizeof(payload)) != 0 || + read(data_pipe[0], &extra, sizeof(extra)) != 0) { + errno = EIO; + return -1; + } + + return close(data_pipe[0]) | close(carrier[0]) | close(carrier[1]); +} + +static int abandoned_endpoint_is_released(void) { + int carrier[2]; + int data_pipe[2]; + if (socketpair(AF_UNIX, SOCK_STREAM, 0, carrier) < 0 || pipe(data_pipe) < 0) + return -1; + if (send_fd(carrier[0], data_pipe[0]) < 0 || close(data_pipe[0]) < 0 || + write(data_pipe[1], "held", 4) != 4 || close(carrier[0]) < 0 || + close(carrier[1]) < 0) { + return -1; + } + + errno = 0; + if (write(data_pipe[1], "released", 8) != -1 || errno != EPIPE) + return -1; + return close(data_pipe[1]); +} + +static int unreturnable_endpoint_is_released(void) { + int carrier[2]; + int data_pipe[2]; + if (socketpair(AF_UNIX, SOCK_STREAM, 0, carrier) < 0 || pipe(data_pipe) < 0) + return -1; + if (send_fd(carrier[0], data_pipe[0]) < 0 || close(data_pipe[0]) < 0 || + write(data_pipe[1], "held", 4) != 4) { + return -1; + } + + char byte = 0; + struct iovec iov = { + .iov_base = &byte, + .iov_len = sizeof(byte), + }; + struct msghdr message = { + .msg_iov = &iov, + .msg_iovlen = 1, + }; + if (recvmsg(carrier[1], &message, 0) != 1 || + (message.msg_flags & MSG_CTRUNC) == 0) { + return -1; + } + + errno = 0; + if (write(data_pipe[1], "released", 8) != -1 || errno != EPIPE) + return -1; + return close(data_pipe[1]) | close(carrier[0]) | close(carrier[1]); +} + +static int reachable_socket_right_cycles_deliver(void) { + int self[2]; + if (socketpair(AF_UNIX, SOCK_STREAM, 0, self) < 0 || + send_fd(self[0], self[1]) < 0) { + return -1; + } + int received_self = receive_fd(self[1]); + char byte = 0; + if (received_self < 0 || close(self[1]) < 0 || + write(self[0], "S", 1) != 1 || read(received_self, &byte, 1) != 1 || + byte != 'S' || write(received_self, "T", 1) != 1 || + read(self[0], &byte, 1) != 1 || byte != 'T' || close(self[0]) < 0 || + close(received_self) < 0) { + return -1; + } + + int a[2]; + int b[2]; + if (socketpair(AF_UNIX, SOCK_STREAM, 0, a) < 0 || + socketpair(AF_UNIX, SOCK_STREAM, 0, b) < 0 || + send_fd(a[0], b[1]) < 0 || send_fd(b[0], a[1]) < 0) { + return -1; + } + int received_b = receive_fd(a[1]); + int received_a = receive_fd(b[1]); + if (received_a < 0 || received_b < 0 || close(a[1]) < 0 || + close(b[1]) < 0 || write(a[0], "A", 1) != 1 || + read(received_a, &byte, 1) != 1 || byte != 'A' || + write(b[0], "B", 1) != 1 || read(received_b, &byte, 1) != 1 || + byte != 'B' || close(a[0]) < 0 || close(b[0]) < 0 || + close(received_a) < 0 || close(received_b) < 0) { + return -1; + } + return 0; +} + +static int abandoned_socket_right_cycles_are_collected(void) { + for (int i = 0; i < 64; ++i) { + int self[2]; + if (socketpair(AF_UNIX, SOCK_STREAM, 0, self) < 0 || + send_fd(self[0], self[1]) < 0 || close(self[0]) < 0 || + close(self[1]) < 0) { + return -1; + } + } + + int a[2]; + int b[2]; + if (socketpair(AF_UNIX, SOCK_STREAM, 0, a) < 0 || + socketpair(AF_UNIX, SOCK_STREAM, 0, b) < 0 || + send_fd(a[0], b[1]) < 0 || send_fd(b[0], a[1]) < 0 || + close(a[0]) < 0 || close(a[1]) < 0 || close(b[0]) < 0 || + close(b[1]) < 0) { + return -1; + } + return 0; +} + +static int receiver_control_truncation_installs_prefix_and_releases_excess(void) { + int carrier[2]; + int first[2]; + int second[2]; + if (socketpair(AF_UNIX, SOCK_STREAM, 0, carrier) < 0 || pipe(first) < 0 || + pipe(second) < 0) { + return -1; + } + + int sent[2] = {first[0], second[0]}; + if (send_fds(carrier[0], sent, 2) < 0 || close(first[0]) < 0 || + close(second[0]) < 0) { + return -1; + } + + int received = -1; + int message_flags = 0; + int count = receive_fds_with_flags(carrier[1], &received, 1, + &message_flags); + if (count != 1 || (message_flags & MSG_CTRUNC) == 0 || + write(first[1], "kept", 4) != 4) { + return -1; + } + + char payload[4]; + if (read(received, payload, sizeof(payload)) != (ssize_t)sizeof(payload) || + memcmp(payload, "kept", sizeof(payload)) != 0) { + errno = EIO; + return -1; + } + + errno = 0; + if (write(second[1], "dropped", 7) != -1 || errno != EPIPE) + return -1; + return close(received) | close(first[1]) | close(second[1]) | + close(carrier[0]) | close(carrier[1]); +} + +static int receiver_emfile_partially_installs_and_releases(void) { + int carrier[2]; + int first[2]; + int second[2]; + if (socketpair(AF_UNIX, SOCK_STREAM, 0, carrier) < 0 || pipe(first) < 0 || + pipe(second) < 0) { + return -1; + } + + int sent[2] = {first[0], second[0]}; + if (send_fds(carrier[0], sent, 2) < 0 || close(first[0]) < 0 || + close(second[0]) < 0) { + return -1; + } + + struct rlimit saved; + if (getrlimit(RLIMIT_NOFILE, &saved) < 0) + return -1; + struct rlimit limited = saved; + limited.rlim_cur = (rlim_t)sent[0] + 1; + if (setrlimit(RLIMIT_NOFILE, &limited) < 0) + return -1; + + int received[2] = {-1, -1}; + int message_flags = 0; + int count = receive_fds_with_flags(carrier[1], received, 2, + &message_flags); + int receive_errno = errno; + if (setrlimit(RLIMIT_NOFILE, &saved) < 0) + return -1; + if (count != 1 || (message_flags & MSG_CTRUNC) == 0) { + errno = receive_errno ? receive_errno : EMFILE; + return -1; + } + + if (write(first[1], "kept", 4) != 4) + return -1; + char payload[4]; + if (read(received[0], payload, sizeof(payload)) != (ssize_t)sizeof(payload) || + memcmp(payload, "kept", sizeof(payload)) != 0) { + errno = EIO; + return -1; + } + + errno = 0; + if (write(second[1], "dropped", 7) != -1 || errno != EPIPE) + return -1; + return close(received[0]) | close(first[1]) | close(second[1]) | + close(carrier[0]) | close(carrier[1]); +} + +int main(void) { + if (signal(SIGPIPE, SIG_IGN) == SIG_ERR || + transferred_endpoint_survives_sender_close() < 0 || + transferred_writer_survives_sender_close() < 0 || + abandoned_endpoint_is_released() < 0 || + unreturnable_endpoint_is_released() < 0 || + reachable_socket_right_cycles_deliver() < 0 || + abandoned_socket_right_cycles_are_collected() < 0 || + receiver_control_truncation_installs_prefix_and_releases_excess() < 0 || + receiver_emfile_partially_installs_and_releases() < 0) { + perror("scm-rights-pipe-lifetime"); + return 1; + } + puts("PASS: SCM_RIGHTS owns pipe endpoints in flight and after receipt"); + return 0; +}