From 9189c4c6dc38eaba4af5b3de65d1e980797f557a Mon Sep 17 00:00:00 2001 From: Ibuki Omatsu Date: Sat, 19 Jul 2025 03:58:53 +0000 Subject: [PATCH] feat: Implement uds_dgram and uds_stream schemes. --- Cargo.lock | 59 ++- bootstrap/Cargo.lock | 4 +- bootstrap/src/procmgr.rs | 62 ++- ipcd/Cargo.toml | 5 +- ipcd/src/main.rs | 103 +++- ipcd/src/uds/dgram.rs | 613 ++++++++++++++++++++++ ipcd/src/uds/mod.rs | 395 +++++++++++++++ ipcd/src/uds/stream.rs | 1035 ++++++++++++++++++++++++++++++++++++++ 8 files changed, 2247 insertions(+), 29 deletions(-) create mode 100644 ipcd/src/uds/dgram.rs create mode 100644 ipcd/src/uds/mod.rs create mode 100644 ipcd/src/uds/stream.rs diff --git a/Cargo.lock b/Cargo.lock index e52c374963..53f3894279 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -358,6 +358,11 @@ dependencies = [ "typenum", ] +[[package]] +name = "generic-rt" +version = "0.1.0" +source = "git+https://gitlab.redox-os.org/redox-os/relibc.git#518798f64e313a2d9067a52b6a3d5970c8208ae5" + [[package]] name = "getrandom" version = "0.2.16" @@ -369,6 +374,17 @@ dependencies = [ "wasi 0.11.0+wasi-snapshot-preview1", ] +[[package]] +name = "goblin" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f27c1b4369c2cd341b5de549380158b105a04c331be5db9110eef7b6d2742134" +dependencies = [ + "log", + "plain", + "scroll", +] + [[package]] name = "hash32" version = "0.3.1" @@ -476,8 +492,11 @@ name = "ipcd" version = "0.1.0" dependencies = [ "anyhow", + "libc", "libredox", + "log", "redox-daemon", + "redox-rt", "redox-scheme 0.7.0", "redox_event", "redox_syscall", @@ -501,15 +520,15 @@ dependencies = [ [[package]] name = "libc" -version = "0.2.171" +version = "0.2.174" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c19937216e9d3aa9956d9bb8dfc0b0c8beb6058fc4f7a4dc4d850edf86a237d6" +checksum = "1171693293099992e19cddea4e8b849964e9846f4acee11b3948bcc337be8776" [[package]] name = "libredox" -version = "0.1.3" +version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c0ff37bd590ca25063e35af745c343cb7a0271906fb7b37e4813e8f79f00268d" +checksum = "1580801010e535496706ba011c15f8532df6b42297d2e471fec38ceadd8c0638" dependencies = [ "bitflags 2.9.0", "ioslice", @@ -836,6 +855,18 @@ dependencies = [ "termion", ] +[[package]] +name = "redox-rt" +version = "0.1.0" +source = "git+https://gitlab.redox-os.org/redox-os/relibc.git#518798f64e313a2d9067a52b6a3d5970c8208ae5" +dependencies = [ + "bitflags 2.9.0", + "generic-rt", + "goblin", + "plain", + "redox_syscall", +] + [[package]] name = "redox-scheme" version = "0.2.4" @@ -961,6 +992,26 @@ version = "1.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eded382c5f5f786b989652c49544c4877d9f015cc22e145a5ea8ea66c2921cd2" +[[package]] +name = "scroll" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "04c565b551bafbef4157586fa379538366e4385d42082f255bfd96e4fe8519da" +dependencies = [ + "scroll_derive", +] + +[[package]] +name = "scroll_derive" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1db149f81d46d2deba7cd3c50772474707729550221e69588478ebf9ada425ae" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "sha2" version = "0.8.2" diff --git a/bootstrap/Cargo.lock b/bootstrap/Cargo.lock index 75cb245758..132af8510b 100644 --- a/bootstrap/Cargo.lock +++ b/bootstrap/Cargo.lock @@ -46,7 +46,7 @@ checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" [[package]] name = "generic-rt" version = "0.1.0" -source = "git+https://gitlab.redox-os.org/redox-os/relibc.git#696ae4eca1df5e3b76a57f62903ee1a94b9c6aa6" +source = "git+https://gitlab.redox-os.org/redox-os/relibc.git#518798f64e313a2d9067a52b6a3d5970c8208ae5" [[package]] name = "goblin" @@ -150,7 +150,7 @@ checksum = "436d45c2b6a5b159d43da708e62b25be3a4a3d5550d654b72216ade4c4bfd717" [[package]] name = "redox-rt" version = "0.1.0" -source = "git+https://gitlab.redox-os.org/redox-os/relibc.git#696ae4eca1df5e3b76a57f62903ee1a94b9c6aa6" +source = "git+https://gitlab.redox-os.org/redox-os/relibc.git#518798f64e313a2d9067a52b6a3d5970c8208ae5" dependencies = [ "bitflags", "generic-rt", diff --git a/bootstrap/src/procmgr.rs b/bootstrap/src/procmgr.rs index 1cfe6ca309..ba201102f7 100644 --- a/bootstrap/src/procmgr.rs +++ b/bootstrap/src/procmgr.rs @@ -794,29 +794,30 @@ impl<'a> ProcScheme<'a> { _ => return Err(Error::new(ENOENT)), }) } + fn read_process_metadata(&self, pid: ProcessId, buf: &mut [u8]) -> Result { + let proc_rc = self.processes.get(&pid).ok_or(Error::new(ESRCH))?; + let process = proc_rc.borrow(); + let metadata = ProcMeta { + pid: pid.0 as u32, + pgid: process.pgid.0 as u32, + ppid: process.ppid.0 as u32, + ruid: process.ruid, + euid: process.euid, + suid: process.suid, + rgid: process.rgid, + egid: process.egid, + sgid: process.sgid, + ens: process.ens, + rns: process.rns, + }; + *buf.get_mut(..size_of::()) + .and_then(|b| plain::from_mut_bytes(b).ok()) + .ok_or(Error::new(EBADF))? = metadata; + Ok(size_of::()) + } fn on_read(&mut self, id: usize, offset: u64, buf: &mut [u8]) -> Result { match self.handles[id] { - Handle::Proc(pid) => { - let proc_rc = self.processes.get(&pid).ok_or(Error::new(EBADFD))?; - let process = proc_rc.borrow(); - let metadata = ProcMeta { - pid: pid.0 as u32, - pgid: process.pgid.0 as u32, - ppid: process.ppid.0 as u32, - ruid: process.ruid, - euid: process.euid, - suid: process.suid, - rgid: process.rgid, - egid: process.egid, - sgid: process.sgid, - ens: process.ens, - rns: process.rns, - }; - *buf.get_mut(..size_of::()) - .and_then(|b| plain::from_mut_bytes(b).ok()) - .ok_or(Error::new(EINVAL))? = metadata; - Ok(size_of::()) - } + Handle::Proc(pid) => self.read_process_metadata(pid, buf), Handle::Ps(ref src_buf) => { let src_buf = usize::try_from(offset) .ok() @@ -907,6 +908,7 @@ impl<'a> ProcScheme<'a> { } Handle::Proc(fd_pid) => { let Some(verb) = ProcCall::try_from_raw(metadata[0] as usize) else { + log::info!("Invalid proc call: {metadata:?}"); return Response::ready_err(EINVAL, op); }; fn cvt_u32(u: u32) -> Option { @@ -1040,6 +1042,24 @@ impl<'a> ProcScheme<'a> { Response::ready_err(ESRCH, op) } } + ProcCall::GetProcCredentials => { + match self.processes.get(&fd_pid) { + None => { + log::warn!("GetProcCredentials: process {fd_pid:?} does not exist"); + return Response::ready_err(ESRCH, op); + } + Some(proc) if proc.borrow().euid != 0 => { + log::warn!("GetProcCredentials: non-root process {fd_pid:?} tried to get credentials"); + return Response::ready_err(EPERM, op); + } + _ => {} + } + let target_pid = metadata[1] as usize; + Ready(Response::new( + self.read_process_metadata(ProcessId(target_pid), payload), + op, + )) + } } } Handle::Ps(_) => Response::ready_err(EOPNOTSUPP, op), diff --git a/ipcd/Cargo.toml b/ipcd/Cargo.toml index 3637b17368..a0f945b200 100644 --- a/ipcd/Cargo.toml +++ b/ipcd/Cargo.toml @@ -4,9 +4,12 @@ name = "ipcd" version = "0.1.0" [dependencies] -libredox = "0.1.3" +libredox = { version = "0.1.4", features = ["call"]} redox-daemon = "0.1.2" redox_syscall = "0.5" redox-scheme = "0.7.0" +redox-rt = { git = "https://gitlab.redox-os.org/redox-os/relibc.git", default-features = false } redox_event = "0.4.0" anyhow = "1.0.97" +libc = "0.2" +log = "0.4" diff --git a/ipcd/src/main.rs b/ipcd/src/main.rs index 8aaf664fd2..ae7a400b0f 100644 --- a/ipcd/src/main.rs +++ b/ipcd/src/main.rs @@ -6,9 +6,12 @@ use std::sync::Mutex; mod chan; mod shm; +mod uds; use self::chan::ChanScheme; use self::shm::ShmScheme; +use self::uds::dgram::UdsDgramScheme; +use self::uds::stream::UdsStreamScheme; fn main() { redox_daemon::Daemon::new(move |daemon| { @@ -29,6 +32,8 @@ fn inner(daemon: redox_daemon::Daemon) -> anyhow::Result<()> { enum EventSource { ChanSocket, ShmSocket, + UdsStreamSocket, + UdsDgramSocket, } } @@ -44,6 +49,20 @@ fn inner(daemon: redox_daemon::Daemon) -> anyhow::Result<()> { let shm = Mutex::new(ShmScheme::new()); let mut shm_handler = ReadinessBased::new(&shm_socket, 16); + // Prepare uds stream scheme + let uds_stream_socket = Socket::nonblock("uds_stream") + .map_err(|e| anyhow::anyhow!("failed to create uds stream scheme: {e}"))?; + let uds_stream = UdsStreamScheme::new(&uds_stream_socket); + let mut uds_stream_handler = ReadinessBased::new(&uds_stream_socket, 16); + let uds_stream_mutex = Mutex::new(uds_stream); + + // Prepare uds dgram scheme + let uds_dgram_socket = Socket::nonblock("uds_dgram") + .map_err(|e| anyhow::anyhow!("failed to create uds dgram scheme: {e}"))?; + let uds_dgram = UdsDgramScheme::new(&uds_dgram_socket); + let mut uds_dgram_handler = ReadinessBased::new(&uds_dgram_socket, 16); + let uds_dgram_mutex = Mutex::new(uds_dgram); + daemon.ready().unwrap(); // Create event listener for both files @@ -63,13 +82,29 @@ fn inner(daemon: redox_daemon::Daemon) -> anyhow::Result<()> { EventFlags::READ, ) .map_err(|e| anyhow::anyhow!("failed to subscribe shm socket: {e}"))?; + event_queue + .subscribe( + uds_stream_socket.inner().raw(), + EventSource::UdsStreamSocket, + EventFlags::READ, + ) + .map_err(|e| anyhow::anyhow!("failed to subscribe uds stream socket: {e}"))?; + event_queue + .subscribe( + uds_dgram_socket.inner().raw(), + EventSource::UdsDgramSocket, + EventFlags::READ, + ) + .map_err(|e| anyhow::anyhow!("failed to subscribe uds dgram socket: {e}"))?; libredox::call::setrens(0, 0)?; // EOF flags let mut chan_eof = false; let mut shm_eof = false; - while !(chan_eof && shm_eof) { + let mut uds_stream_eof = false; + let mut uds_dgram_eof = false; + while !(chan_eof && shm_eof && uds_stream_eof && uds_dgram_eof) { let Some(event_res) = event_queue.next() else { break; }; @@ -139,6 +174,72 @@ fn inner(daemon: redox_daemon::Daemon) -> anyhow::Result<()> { Err(err) => return Err(anyhow::anyhow!("{err}")), } } + EventSource::UdsStreamSocket => { + // Unix Domain Socket Stream scheme + if !uds_stream_eof { + // 1. Read requests + match uds_stream_handler.read_requests() { + Ok(true) => {} // Read requests success + Ok(false) => { + // EOF + uds_stream_eof = true; + } + Err(err) => return Err(anyhow::anyhow!("{err}")), + } + } + + // 2. Process requests + uds_stream_handler.process_requests(|| uds_stream_mutex.lock().unwrap()); + + // 3.Poll all blocking requests + uds_stream_handler + .poll_all_requests(|| uds_stream_mutex.lock().unwrap()) + .map_err(|e| anyhow::anyhow!("error occured in poll_all_requests: {e}"))?; + + // 3. Write responses + // write_responses returns a Result, but currently only returns true. + match uds_stream_handler.write_responses() { + Ok(true) => {} // Read requests success + Ok(false) => { + // EOF + uds_stream_eof = true; + } + Err(err) => return Err(anyhow::anyhow!("{err}")), + } + } + EventSource::UdsDgramSocket => { + // Unix Domain Socket Dgram scheme + if !uds_dgram_eof { + // 1. Read requests + match uds_dgram_handler.read_requests() { + Ok(true) => {} // Read requests success + Ok(false) => { + // EOF + uds_dgram_eof = true; + } + Err(err) => return Err(anyhow::anyhow!("{err}")), + } + } + + // 2. Process requests + uds_dgram_handler.process_requests(|| uds_dgram_mutex.lock().unwrap()); + + // 3.Poll all blocking requests + uds_dgram_handler + .poll_all_requests(|| uds_dgram_mutex.lock().unwrap()) + .map_err(|e| anyhow::anyhow!("error occured in poll_all_requests: {e}"))?; + + // 3. Write responses + // write_responses returns a Result, but currently only returns true. + match uds_dgram_handler.write_responses() { + Ok(true) => {} // Read requests success + Ok(false) => { + // EOF + uds_dgram_eof = true; + } + Err(err) => return Err(anyhow::anyhow!("{err}")), + } + } } } Ok(()) diff --git a/ipcd/src/uds/dgram.rs b/ipcd/src/uds/dgram.rs new file mode 100644 index 0000000000..94174192d4 --- /dev/null +++ b/ipcd/src/uds/dgram.rs @@ -0,0 +1,613 @@ +//! uds scheme for handling Unix Domain Socket datagram communication + +use super::{ + path_buf_to_string, read_msghdr_info, read_num, AncillaryData, Credential, DataPacket, + MsgWriter, MAX_DGRAM_MSG_LEN, +}; + +use libc::{AF_UNIX, SO_DOMAIN, SO_PASSCRED}; +use redox_rt::protocol::SocketCall; +use redox_scheme::{ + scheme::SchemeSync, CallerCtx, OpenResult, Response, SendFdRequest, SignalBehavior, + Socket as SchemeSocket, +}; +use std::{ + cell::RefCell, + cmp, + collections::{HashMap, HashSet, VecDeque}, + mem, + rc::Rc, +}; +use syscall::{error::*, flag::*, schemev2::NewFdFlags, Error, FobtainFdFlags}; + +impl DataPacket { + pub fn serialize_to_stream( + self, + stream: &mut [u8], + socket: &mut Socket, + name_buf_size: usize, + iov_size: usize, + ) -> Result { + let mut msg_writer = MsgWriter::new(stream); + msg_writer.write_name( + self.ancillary_data.name, + name_buf_size, + UdsDgramScheme::fpath_inner, + )?; + + msg_writer.write_payload(&self.payload, self.payload.len(), iov_size)?; + + // Write the ancillary data + if !msg_writer.write_rights(self.ancillary_data.num_fds) { + // Buffer was too small, FDs could not be described. Drop the actual FDs. + log::warn!( + "serialize_to_stream: Buffer too small for SCM_RIGHTS, dropping {} FDs.", + self.ancillary_data.num_fds + ); + socket.drop_fds(self.ancillary_data.num_fds)?; + } + // Write other ancillary datas + for option in &socket.options { + let result = match *option { + SO_PASSCRED => msg_writer.write_credentials(&self.ancillary_data.cred), + _ => { + log::warn!( + "serialize_to_stream: Unsupported socket option for serialization: {}", + option + ); + return Err(Error::new(EOPNOTSUPP)); + } + }; + if !result { + log::warn!("serialize_to_stream: Buffer too small for ancillary data, stopping further serialization."); + break; + } + } + + Ok(msg_writer.len()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum State { + Unbound, + Bound, + Closed, +} + +impl Default for State { + fn default() -> Self { + Self::Unbound + } +} + +#[derive(Debug, Default)] +pub struct Socket { + primary_id: usize, + path: Option, + state: State, + peer: Option, + messages: VecDeque, + options: HashSet, + fds: VecDeque, + flags: usize, +} + +impl Socket { + fn drop_fds(&mut self, num_fd: usize) -> Result<()> { + for i in 0..num_fd { + if self.fds.pop_front().is_none() { + log::error!("Socket::drop_fds: Attempted to drop FD #{} of {}, but fd queue is empty. State inconsistency.", i + 1, num_fd); + return Err(Error::new(EINVAL)); + } + } + Ok(()) + } +} + +pub struct UdsDgramScheme<'sock> { + sockets: HashMap>>, + next_id: usize, + socket_paths: HashMap>>, + socket: &'sock SchemeSocket, +} + +impl<'sock> UdsDgramScheme<'sock> { + pub fn new(socket: &'sock SchemeSocket) -> Self { + Self { + sockets: HashMap::new(), + next_id: 0, + socket_paths: HashMap::new(), + socket, + } + } + + fn post_fevent(&self, id: usize, flags: usize) -> Result<()> { + let fevent_response = Response::post_fevent(id, flags); + match self + .socket + .write_response(fevent_response, SignalBehavior::Restart) + { + Ok(true) => Ok(()), + Ok(false) => Err(Error::new(EAGAIN)), + Err(err) => Err(err), + } + } + + fn get_socket(&self, id: usize) -> Result<&Rc>, Error> { + self.sockets.get(&id).ok_or(Error::new(EBADF)) + } + + fn get_connected_peer(&self, id: usize) -> Result<(usize, Rc>), Error> { + let socket = self.get_socket(id)?.borrow(); + + let remote_id = socket.peer.ok_or(Error::new(ENOTCONN))?; + + let remote_rc = self.get_socket(remote_id).map_err(|e| { + log::error!("get_connected_peer(id: {}): Peer socket (id: {}) has vanished. Original error: {:?}", id, remote_id, e); + Error::new(EPIPE) + })?; + + if remote_rc.borrow().state == State::Closed { + log::warn!( + "get_connected_peer(id: {}): Attempted to interact with a closed peer (id: {}).", + id, + remote_id + ); + return Err(Error::new(ECONNREFUSED)); + } + + Ok((remote_id, remote_rc.clone())) + } + + fn handle_unnamed_socket(&mut self, flags: usize) -> usize { + let new_id = self.next_id; + let mut new = Socket::default(); + new.flags = flags; + new.primary_id = new_id; + + self.sockets.insert(new_id, Rc::new(RefCell::new(new))); + self.next_id += 1; + new_id + } + + fn call_inner( + &mut self, + id: usize, + payload: &mut [u8], + metadata: &[u64], + ctx: &CallerCtx, + ) -> Result { + // metadata to Vec + let Some(verb) = SocketCall::try_from_raw(metadata[0] as usize) else { + log::error!("call_inner: Invalid verb in metadata: {:?}", metadata); + return Err(Error::new(EINVAL)); + }; + match verb { + SocketCall::Bind => self.handle_bind(id, &payload), + SocketCall::Connect => self.handle_connect(id, &payload), + SocketCall::SetSockOpt => self.handle_setsockopt(id, metadata[1] as i32, &payload), + SocketCall::GetSockOpt => self.handle_getsockopt(id, metadata[1] as i32, payload), + SocketCall::SendMsg => self.handle_sendmsg(id, payload, ctx), + SocketCall::RecvMsg => self.handle_recvmsg(id, payload), + _ => Err(Error::new(OPNOTSUP)), + } + } + + fn handle_bind(&mut self, id: usize, path_buf: &[u8]) -> Result { + let path = path_buf_to_string(path_buf)?; + + // Check if path is already bound to a server + if self.socket_paths.contains_key(&path) { + log::warn!( + "handle_bind(id: {}): Address '{}' already in use.", + id, + path + ); + return Err(Error::new(EADDRINUSE)); + } + + let socket_rc = self.get_socket(id)?.clone(); + let mut socket = socket_rc.borrow_mut(); + + if socket.state != State::Unbound { + log::error!( + "handle_bind(id: {}): Socket is already bound or connected (state: {:?})", + id, + socket.state + ); + return Err(Error::new(EINVAL)); + } + + socket.path = Some(path.clone()); + socket.state = State::Bound; + + self.socket_paths.insert(path, socket_rc.clone()); + + Ok(0) + } + + fn handle_connect(&mut self, id: usize, path_buf: &[u8]) -> Result { + let path = path_buf_to_string(path_buf)?; + + let target_rc = self + .socket_paths + .get(&path) + .ok_or(Error::new(ECONNREFUSED))?; + let target_id = target_rc.borrow().primary_id; + + let socket_rc = self.get_socket(id)?; + socket_rc.borrow_mut().peer = Some(target_id); + + Ok(0) + } + + fn handle_setsockopt(&mut self, id: usize, option: i32, value_slice: &[u8]) -> Result { + let socket_rc = self.get_socket(id)?; + let mut socket = socket_rc.borrow_mut(); + + match option { + SO_PASSCRED => { + let value = read_num::(value_slice)?; + if value != 0 { + socket.options.insert(SO_PASSCRED); + } else { + socket.options.remove(&SO_PASSCRED); + } + Ok(value_slice.len()) + } + _ => { + log::warn!( + "handle_setsockopt(id: {}): Unsupported option: {}", + id, + option + ); + Err(Error::new(ENOPROTOOPT)) + } + } + } + + fn handle_getsockopt(&mut self, id: usize, option: i32, payload: &mut [u8]) -> Result { + match option { + SO_DOMAIN => { + payload.fill(0); + if payload.len() < mem::size_of::() { + log::error!( + "handle_getsockopt(id: {}): SO_DOMAIN payload buffer is too small. len: {}", + id, + payload.len() + ); + return Err(Error::new(ENOBUFS)); + } + let domain = AF_UNIX.to_le_bytes(); + payload[..domain.len()].copy_from_slice(&domain); + Ok(domain.len()) + } + _ => { + log::warn!( + "handle_getsockopt(id: {}): Unsupported option: {}", + id, + option + ); + Err(Error::new(ENOPROTOOPT)) + } + } + } + + fn handle_sendmsg(&mut self, id: usize, msg_stream: &[u8], ctx: &CallerCtx) -> Result { + if msg_stream.is_empty() { + log::error!("handle_sendmsg(id: {}): msg_stream is empty.", id); + return Err(Error::new(EINVAL)); + } + + let name = { + let socket_rc = self.get_socket(id)?; + let socket = socket_rc.borrow(); + socket.path.clone() + }; + let (remote_id, remote_rc) = self.get_connected_peer(id)?; + + let bytes_written = + Self::sendmsg_inner(&mut remote_rc.borrow_mut(), name, msg_stream, ctx)?; + self.post_fevent(remote_id, EVENT_READ.bits())?; + Ok(bytes_written) + } + + fn sendmsg_inner( + socket: &mut Socket, + name: Option, + msg_stream: &[u8], + ctx: &CallerCtx, + ) -> Result { + if msg_stream.is_empty() { + log::error!("sendmsg_inner: msg_stream is empty."); + return Err(Error::new(EINVAL)); + } + + let message = DataPacket::from_stream(msg_stream, name, ctx)?; + let payload_len = message.len(); + socket.messages.push_back(message); + + Ok(payload_len) + } + + fn handle_recvmsg(&mut self, id: usize, msg_stream: &mut [u8]) -> Result { + let socket_rc = self.get_socket(id)?; + let mut socket = socket_rc.borrow_mut(); + + if let Some(message) = socket.messages.pop_front() { + Ok(Self::recvmsg_inner(&mut socket, message, msg_stream)?) + } else if (socket.flags as usize) & O_NONBLOCK == O_NONBLOCK { + Err(Error::new(EAGAIN)) + } else { + Err(Error::new(EWOULDBLOCK)) + } + } + + fn recvmsg_inner( + socket: &mut Socket, + message: DataPacket, + msg_stream: &mut [u8], + ) -> Result { + // Read the name length, whole iov size, and msg controllen from the stream + let (prepared_name_len, prepared_whole_iov_size, _) = read_msghdr_info(msg_stream)?; + + message.serialize_to_stream( + msg_stream, + socket, + prepared_name_len, + prepared_whole_iov_size, + ) + } + + fn handle_connect_socketpair(&mut self, id: usize) -> Result { + let new_id = self.next_id; + let mut new = Socket::default(); + new.primary_id = new_id; + + let socket_rc = self.get_socket(id)?; + if socket_rc.borrow().state == State::Closed { + log::warn!( + "handle_connect_socketpair(id: {}): Attempting to connect from a closed socket.", + id + ); + return Err(Error::new(ECONNREFUSED)); + } + + { + let mut socket = socket_rc.borrow_mut(); + socket.peer = Some(new_id); + } + + new.peer = Some(id); + + // smoltcp sends writeable whenever a listener gets a + // client, we'll do the same too (but also readable, + // why not) + self.post_fevent(id, (EVENT_READ | EVENT_WRITE).bits())?; + + self.sockets.insert(new_id, Rc::new(RefCell::new(new))); + + self.next_id += 1; + + Ok(OpenResult::ThisScheme { + number: new_id, + flags: NewFdFlags::empty(), + }) + } + + fn handle_recvfd(&mut self, id: usize) -> Result { + let socket_rc = self.get_socket(id)?; + let mut socket = socket_rc.borrow_mut(); + let fd = socket.fds.pop_front().ok_or(Error::new(EWOULDBLOCK))?; + + Ok(OpenResult::OtherScheme { fd }) + } + + fn handle_listen(&mut self, id: usize) -> Result { + let socket_rc = self.get_socket(id)?; + + let new_id = self.next_id; + + self.sockets.insert(new_id, socket_rc.clone()); + self.next_id += 1; + + Ok(OpenResult::ThisScheme { + number: new_id, + flags: NewFdFlags::empty(), + }) + } + + fn write_inner(&mut self, id: usize, buf: &[u8], ctx: &CallerCtx) -> Result { + if buf.len() > MAX_DGRAM_MSG_LEN { + return Err(Error::new(EMSGSIZE)); + } + + let name = { + let socket_rc = self.get_socket(id)?; + let socket = socket_rc.borrow(); + if matches!(socket.state, State::Closed) { + return Err(Error::new(EPIPE)); + } + + socket.path.clone() + }; + + // Assume writing to the connected socket if the given id is the primary id + let (remote_id, remote_rc) = self.get_connected_peer(id)?; + let mut remote = remote_rc.borrow_mut(); + let message = DataPacket::new( + buf.to_vec(), + AncillaryData::new( + Credential::new(ctx.pid as i32, ctx.uid as i32, ctx.gid as i32), + name, + ), + ); + remote.messages.push_back(message); + + self.post_fevent(remote_id, EVENT_READ.bits())?; + + Ok(buf.len()) + } + + fn fpath_inner(path: &String, buf: &mut [u8]) -> Result { + // Write scheme name + const PREFIX: &[u8] = b"/scheme/uds_dgram/"; + let len = cmp::min(PREFIX.len(), buf.len()); + buf[..len].copy_from_slice(&PREFIX[..len]); + if len < PREFIX.len() { + return Ok(len); + } + + // Write path + let len = cmp::min(path.len(), buf.len() - PREFIX.len()); + buf[PREFIX.len()..][..len].copy_from_slice(&path.as_bytes()[..len]); + + Ok(PREFIX.len() + len) + } + + fn read_inner(&mut self, id: usize, buf: &mut [u8], flags: u32) -> Result { + let socket_rc = self.get_socket(id)?; + let mut socket = socket_rc.borrow_mut(); + + if let Some(message) = socket.messages.pop_front() { + let full_len = message.len(); + let copy_len = cmp::min(buf.len(), full_len); + buf[..copy_len].copy_from_slice(&message.payload[..copy_len]); + + Ok(copy_len) + } else if (flags as usize) & O_NONBLOCK == O_NONBLOCK { + Err(Error::new(EAGAIN)) + } else { + Err(Error::new(EWOULDBLOCK)) + } + } + + fn sendfd_inner(&mut self, sendfd_request: &SendFdRequest) -> Result { + let mut new_fd = usize::MAX; + if let Err(e) = + sendfd_request.obtain_fd(&self.socket, FobtainFdFlags::empty(), Err(&mut new_fd)) + { + log::error!("sendfd_inner: obtain_fd failed with error: {:?}", e); + return Err(e); + } + let socket_id = sendfd_request.id(); + let (remote_id, remote_rc) = self.get_connected_peer(socket_id)?; + remote_rc.borrow_mut().fds.push_back(new_fd); + + self.post_fevent(remote_id, EVENT_READ.bits())?; + Ok(new_fd) + } +} + +impl<'sock> SchemeSync for UdsDgramScheme<'sock> { + fn open(&mut self, path: &str, flags: usize, _ctx: &CallerCtx) -> Result { + let new_id = if path.is_empty() { + self.handle_unnamed_socket(flags) + } else { + log::error!( + "open(path: '{}'): Attempting to open a named socket, which is not supported.", + path + ); + return Err(Error::new(EINVAL)); + }; + Ok(OpenResult::ThisScheme { + number: new_id, + flags: NewFdFlags::empty(), + }) + } + + fn call( + &mut self, + id: usize, + payload: &mut [u8], + metadata: &[u64], + ctx: &CallerCtx, + ) -> Result { + self.call_inner(id, payload, metadata, ctx) + } + + fn dup(&mut self, id: usize, buf: &[u8], _ctx: &CallerCtx) -> Result { + match buf { + // Connect for socket pair + b"connect" => self.handle_connect_socketpair(id), + b"recvfd" => self.handle_recvfd(id), + // listen will generate a id for same socket + b"listen" => self.handle_listen(id), + _ => Err(Error::new(EINVAL)), + } + } + + fn write( + &mut self, + id: usize, + buf: &[u8], + _offset: u64, + _flags: u32, + ctx: &CallerCtx, + ) -> Result { + self.write_inner(id, buf, ctx) + } + + fn fpath(&mut self, id: usize, buf: &mut [u8], _ctx: &CallerCtx) -> Result { + let socket_rc = self.get_socket(id)?; + let socket = socket_rc.borrow(); + + let path = socket.path.as_ref().ok_or_else(|| { + log::error!("fpath(id: {}): Socket is unnamed, has no path.", id); + Error::new(ENOENT) + })?; + Ok(Self::fpath_inner(path, buf)?) + } + + fn fsync(&mut self, id: usize, _ctx: &CallerCtx) -> Result<()> { + self.get_socket(id).and(Ok(())) + } + + fn read( + &mut self, + id: usize, + buf: &mut [u8], + _offset: u64, + flags: u32, + _ctx: &CallerCtx, + ) -> Result { + self.read_inner(id, buf, flags) + } + + fn on_close(&mut self, id: usize) { + let Some(socket_rc) = self.sockets.remove(&id) else { + return; + }; + let mut socket = socket_rc.borrow_mut(); + if socket.primary_id == id { + socket.state = State::Closed; + socket.peer = None; + let path = socket.path.clone(); + socket.path = None; + + if let Some(path) = path { + self.socket_paths.remove(&path); + } + } + } + + fn on_sendfd(&mut self, sendfd_request: &SendFdRequest) -> Result { + self.sendfd_inner(sendfd_request) + } + + fn fcntl(&mut self, id: usize, cmd: usize, arg: usize, _ctx: &CallerCtx) -> Result { + let socket = self.get_socket(id)?; + match cmd { + F_GETFL => Ok(socket.borrow().flags), + F_SETFL => { + socket.borrow_mut().flags = arg; + Ok(0) + } + _ => { + log::error!("fcntl(id: {}): Unsupported cmd: {}", id, cmd); + Err(Error::new(EINVAL)) + } + } + } +} diff --git a/ipcd/src/uds/mod.rs b/ipcd/src/uds/mod.rs new file mode 100644 index 0000000000..bdaa2ea475 --- /dev/null +++ b/ipcd/src/uds/mod.rs @@ -0,0 +1,395 @@ +use redox_rt::protocol::ProcMeta; +use redox_scheme::CallerCtx; +use std::{cmp, convert::TryInto, mem}; +use syscall::{error::*, Error}; + +pub mod dgram; +pub mod stream; + +// TODO: Remove this when Rust libc crate is updated to include SCM_CREDENTIALS +const SCM_CREDENTIALS: i32 = 2; + +const MAX_DGRAM_MSG_LEN: usize = 65536; +const MIN_RECV_MSG_LEN: usize = mem::size_of::() * 2; // name_len, payload_len, +const CMSG_HEADER_LEN_IN_STREAM: usize = CMSG_LEVEL_SIZE + CMSG_TYPE_SIZE + CMSG_DATA_LEN_SIZE; +const CMSG_LEVEL_SIZE: usize = mem::size_of::(); +const CMSG_TYPE_SIZE: usize = mem::size_of::(); +const CMSG_DATA_LEN_SIZE: usize = mem::size_of::(); +const PID_SIZE: usize = mem::size_of::(); +const UID_SIZE: usize = mem::size_of::(); +const GID_SIZE: usize = mem::size_of::(); + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +struct Credential { + pid: i32, + uid: i32, + gid: i32, +} +impl Credential { + fn new(pid: i32, uid: i32, gid: i32) -> Self { + Self { pid, uid, gid } + } +} + +#[derive(Debug, Default, Clone, PartialEq, Eq)] +struct AncillaryData { + cred: Credential, + num_fds: usize, + name: Option, +} +impl AncillaryData { + fn new(cred: Credential, name: Option) -> Self { + Self { + cred, + name, + ..Default::default() + } + } +} + +struct AncillaryDataHeader { + level: i32, + c_type: i32, + data_len: usize, +} +impl AncillaryDataHeader { + fn from_stream(stream: &[u8]) -> Result> { + let mut cursor = 0; + if cursor + CMSG_HEADER_LEN_IN_STREAM > stream.len() { + return Ok(None); + } + + // cmsg entry format: [level(i32)][type(i32)][data_len(usize)][data] + let cmsg_level = read_num::(&stream[cursor..])?; + cursor += mem::size_of::(); + + let cmsg_type = read_num::(&stream[cursor..])?; + cursor += mem::size_of::(); + + let cmsg_data_len = read_num::(&stream[cursor..])?; + + Ok(Some(Self { + level: cmsg_level, + c_type: cmsg_type, + data_len: cmsg_data_len, + })) + } +} + +#[derive(Debug, Default, Clone, PartialEq, Eq)] +struct DataPacket { + payload: Vec, + ancillary_data: AncillaryData, + // Only for stream packets + read_offset: usize, + ancillary_taken: bool, +} + +impl DataPacket { + fn new(payload: Vec, ancillary_data: AncillaryData) -> Self { + Self { + payload, + ancillary_data, + read_offset: 0, + ancillary_taken: false, + } + } + + fn len(&self) -> usize { + self.payload.len() + } + + fn from_stream(stream: &[u8], name: Option, ctx: &CallerCtx) -> Result { + let mut cursor: usize = 0; + let payload_len = read_num::(&stream[cursor..])?; + cursor += mem::size_of::(); + let payload = stream + .get(cursor..cursor + payload_len) + .ok_or_else(|| { + log::error!("Message::from_stream: Stream too short for payload. Expected len {}, actual remaining len {}", payload_len, stream.len() - cursor); + Error::new(EINVAL) + })?; + cursor += payload_len; + + let (pid, uid, gid) = get_uid_gid_from_pid(ctx.pid)?; + + // Create a new message with the payload and credentials + let mut message = Self::new( + payload.to_vec(), + AncillaryData::new(Credential::new(pid as i32, uid as i32, gid as i32), name), + ); + + while let Some(cmsg_header) = AncillaryDataHeader::from_stream(&stream[cursor..])? { + cursor += CMSG_HEADER_LEN_IN_STREAM; + let data_stream = stream + .get(cursor..cursor + cmsg_header.data_len) + .ok_or_else(|| { + log::error!("Message::from_stream: Stream too short for ancillary data. Expected len {}, actual remaining len {}", cmsg_header.data_len, stream.len() - cursor); + Error::new(EINVAL) + })?; + + match (cmsg_header.level, cmsg_header.c_type) { + (libc::SOL_SOCKET, libc::SCM_RIGHTS) => { + // Handle file descriptor passing + let num_fds = read_num::(data_stream)?; + message.ancillary_data.num_fds += num_fds; + } + _ => { + log::warn!( + "Message::from_stream: Unsupported cmsg type received. level: {}, type: {}", + cmsg_header.level, + cmsg_header.c_type + ); + return Err(Error::new(EOPNOTSUPP)); + } + } + cursor += cmsg_header.data_len; + } + Ok(message) + } +} + +fn read_num(buffer: &[u8]) -> Result +where + T: NumFromBytes, +{ + T::from_le_bytes_slice(buffer) +} +trait NumFromBytes: Sized { + fn from_le_bytes_slice(buffer: &[u8]) -> Result; +} +impl NumFromBytes for i32 { + fn from_le_bytes_slice(buffer: &[u8]) -> Result { + Ok(i32::from_le_bytes( + buffer + .get(..mem::size_of::()) + .and_then(|slice| slice.try_into().ok()) + .ok_or_else(|| { + log::error!( + "read_num: buffer is too short to read num len: {}", + buffer.len() + ); + Error::new(EINVAL) + })?, + )) + } +} +impl NumFromBytes for u32 { + fn from_le_bytes_slice(buffer: &[u8]) -> Result { + Ok(u32::from_le_bytes( + buffer + .get(..mem::size_of::()) + .and_then(|slice| slice.try_into().ok()) + .ok_or_else(|| { + log::error!( + "read_num: buffer is too short to read num len: {}", + buffer.len() + ); + Error::new(EINVAL) + })?, + )) + } +} +impl NumFromBytes for usize { + fn from_le_bytes_slice(buffer: &[u8]) -> Result { + Ok(usize::from_le_bytes( + buffer + .get(..mem::size_of::()) + .and_then(|slice| slice.try_into().ok()) + .ok_or_else(|| { + log::error!( + "read_num: buffer is too short to read num len: {}", + buffer.len() + ); + Error::new(EINVAL) + })?, + )) + } +} + +fn get_uid_gid_from_pid(target_pid: usize) -> Result<(u32, u32, u32)> { + let mut buffer = [0u8; mem::size_of::()]; + let _ = libredox::call::get_proc_credentials(target_pid, &mut buffer).map_err(|e| { + log::error!( + "Failed to get process credentials for pid {}: {:?}", + target_pid, + e + ); + Error::new(EINVAL) + })?; + let mut cursor = 0; + let pid = read_num::(&buffer[cursor..])?; + cursor += mem::size_of::() * 3; + let uid = read_num::(&buffer[cursor..])?; + cursor += mem::size_of::() * 3; + let gid = read_num::(&buffer[cursor..])?; + Ok((pid, uid, gid)) +} + +fn read_msghdr_info(stream: &mut [u8]) -> Result<(usize, usize, usize)> { + if stream.len() < mem::size_of::() * 3 { + log::error!( + "get_msghdr_info: stream buffer is too small to read headers. len: {}", + stream.len() + ); + return Err(Error::new(EINVAL)); + } + let mut cursor: usize = 0; + let prepared_name_len = read_num::(&stream[cursor..])?; + cursor += mem::size_of::(); + let prepared_whole_iov_size = read_num::(&stream[cursor..])?; + cursor += mem::size_of::(); + let prepared_msg_controllen = read_num::(&stream[cursor..])?; + cursor += mem::size_of::(); + // Clear the stream buffer + stream[..cursor].copy_from_slice(&[0u8; mem::size_of::() * 3]); + Ok(( + prepared_name_len, + prepared_whole_iov_size, + prepared_msg_controllen, + )) +} + +struct MsgWriter<'a> { + buffer: &'a mut [u8], + written_len: usize, +} +impl<'a> MsgWriter<'a> { + fn new(buffer: &'a mut [u8]) -> Self { + Self { + buffer, + written_len: 0, + } + } + + pub fn len(&self) -> usize { + self.written_len + } + + fn write_name( + &mut self, + name: Option, + name_buf_size: usize, + name_write_fn: fn(&String, &mut [u8]) -> Result, + ) -> Result<()> { + if self.buffer.len() < self.written_len + mem::size_of::() { + log::warn!("MsgWriter::write_name: Buffer too small to write name length. written_len: {}, buffer_len: {}", self.written_len, self.buffer.len()); + self.written_len = self.buffer.len(); + return Ok(()); + } + if let Some(name) = name { + let copy_len = cmp::min( + name_buf_size, + self.buffer.len() - self.written_len - mem::size_of::(), + ); + if copy_len < name.len() { + log::warn!("MsgWriter::write_name: Name will be truncated. Full length: {}, buffer available: {}", name.len(), copy_len); + } + let name_len = name_write_fn( + &name, + &mut self.buffer[self.written_len + mem::size_of::() + ..self.written_len + mem::size_of::() + copy_len], + )?; + self.buffer[self.written_len..self.written_len + mem::size_of::()] + .copy_from_slice(&name_len.to_le_bytes()); + self.written_len += mem::size_of::() + name_len; + Ok(()) + } else { + self.written_len += mem::size_of::(); + Ok(()) + } + } + + fn write_payload(&mut self, payload: &[u8], full_len: usize, iov_size: usize) -> Result { + let Some(payload_len_buffer) = self + .buffer + .get_mut(self.written_len..self.written_len + mem::size_of::()) + else { + log::warn!("MsgWriter::write_payload: Buffer too small to write payload length. written_len: {}, buffer_len: {}", self.written_len, self.buffer.len()); + self.written_len = self.buffer.len(); + return Ok(0); + }; + payload_len_buffer.copy_from_slice(&full_len.to_le_bytes()); + self.written_len += mem::size_of::(); + + let copy_len = cmp::min(iov_size, full_len); + if copy_len < full_len { + log::warn!("MsgWriter::write_payload: Payload will be truncated. Full length: {}, buffer available: {}", full_len, copy_len); + } + let Some(payload_buffer) = self + .buffer + .get_mut(self.written_len..self.written_len + copy_len) + else { + log::warn!("MsgWriter::write_payload: Buffer too small to write payload data. written_len: {}, buffer_len: {}", self.written_len, self.buffer.len()); + self.written_len = self.buffer.len(); + return Ok(0); + }; + payload_buffer.copy_from_slice(&payload[..copy_len]); + self.written_len += copy_len; + Ok(copy_len) + } + + fn write_cmsg(&mut self, level: i32, c_type: i32, data: &[u8]) -> bool { + let data_len = data.len(); + + let Some(remaining_buf) = self.buffer.get_mut(self.written_len..) else { + log::warn!("CmsgWriter::write_cmsg: No remaining buffer space at all."); + return false; + }; + if remaining_buf.len() < CMSG_HEADER_LEN_IN_STREAM { + log::warn!("CmsgWriter::write_cmsg: Not enough space for cmsg header. remaining: {}, needed: {}", remaining_buf.len(), CMSG_HEADER_LEN_IN_STREAM); + // Fill the remaining buffer with 1s to indicate Imcomplete CMSG header + remaining_buf.fill(1); + self.written_len += remaining_buf.len(); + return false; + } + let mut cursor = 0; + remaining_buf[cursor..cursor + CMSG_LEVEL_SIZE].copy_from_slice(&level.to_le_bytes()); + cursor += CMSG_LEVEL_SIZE; + remaining_buf[cursor..cursor + CMSG_TYPE_SIZE].copy_from_slice(&c_type.to_le_bytes()); + cursor += CMSG_TYPE_SIZE; + remaining_buf[cursor..cursor + CMSG_DATA_LEN_SIZE].copy_from_slice(&data_len.to_le_bytes()); + cursor += CMSG_DATA_LEN_SIZE; + + if remaining_buf.len() < cursor + data_len { + log::warn!( + "CmsgWriter::write_cmsg: Not enough space for cmsg data. remaining: {}, needed: {}", + remaining_buf.len(), + cursor + data_len + ); + // Fill the remaining buffer with 1s to indicate Imcomplete CMSG data + remaining_buf.fill(1); + self.written_len += remaining_buf.len(); + return false; + } + remaining_buf[cursor..cursor + data_len].copy_from_slice(data); + self.written_len += cursor + data_len; + + true + } + + fn write_rights(&mut self, num_fds: usize) -> bool { + let data = num_fds.to_le_bytes(); + if num_fds == 0 { + return true; + } + self.write_cmsg(libc::SOL_SOCKET, libc::SCM_RIGHTS, &data) + } + + fn write_credentials(&mut self, credential: &Credential) -> bool { + let mut data = [0u8; PID_SIZE + UID_SIZE + GID_SIZE]; + data[..PID_SIZE].copy_from_slice(&credential.pid.to_le_bytes()); + data[PID_SIZE..PID_SIZE + UID_SIZE].copy_from_slice(&credential.uid.to_le_bytes()); + data[PID_SIZE + UID_SIZE..PID_SIZE + UID_SIZE + GID_SIZE] + .copy_from_slice(&credential.gid.to_le_bytes()); + self.write_cmsg(libc::SOL_SOCKET, SCM_CREDENTIALS, &data) + } +} + +fn path_buf_to_string(path_buf: &[u8]) -> Result { + match std::str::from_utf8(path_buf) { + Ok("") => Err(Error::new(EINVAL)), + Ok(s) => Ok(s.to_string()), + Err(_) => Err(Error::new(EINVAL)), + } +} diff --git a/ipcd/src/uds/stream.rs b/ipcd/src/uds/stream.rs new file mode 100644 index 0000000000..3835bbe37f --- /dev/null +++ b/ipcd/src/uds/stream.rs @@ -0,0 +1,1035 @@ +//! uds scheme for handling Unix Domain Socket stream communication + +use super::{ + path_buf_to_string, read_msghdr_info, read_num, AncillaryData, Credential, DataPacket, + MsgWriter, MIN_RECV_MSG_LEN, +}; + +use libc::{AF_UNIX, SO_DOMAIN, SO_PASSCRED}; +use redox_rt::protocol::SocketCall; +use redox_scheme::{ + scheme::SchemeSync, CallerCtx, OpenResult, Response, SendFdRequest, SignalBehavior, + Socket as SchemeSocket, +}; +use std::{ + cell::RefCell, + cmp, + collections::{HashMap, HashSet, VecDeque}, + mem, + rc::Rc, +}; +use syscall::{error::*, flag::*, schemev2::NewFdFlags, Error}; + +#[derive(Debug, Default, Clone, PartialEq, Eq)] +struct Connection { + peer: usize, + packets: VecDeque, + fds: VecDeque, + + is_peer_shutdown: bool, +} +impl Connection { + fn new(peer: usize) -> Self { + Self { + peer, + ..Default::default() + } + } + + fn drop_fds(&mut self, num_fd: usize) -> Result<()> { + for i in 0..num_fd { + if self.fds.pop_front().is_none() { + log::error!("Connection::drop_fds: Attempted to drop FD #{} of {}, but fd queue is empty. State inconsistency.", i + 1, num_fd); + return Err(Error::new(EPROTO)); + } + } + Ok(()) + } + + fn serialize_to_msgstream( + &mut self, + stream: &mut [u8], + name_buf_size: usize, + iov_size: usize, + options: HashSet, + ) -> Result { + let mut name: Option = None; + let mut payload_buffer: Vec = Vec::with_capacity(iov_size); + let mut ancillary_data_buffer: VecDeque = VecDeque::new(); + let mut total_copied_len = 0; + let mut user_buf_offset = 0; + + while user_buf_offset < iov_size { + let Some(packet) = self.packets.front_mut() else { + // No more packets to read + break; + }; + + let packet_rem_payload = &packet.payload[packet.read_offset..]; + + let user_buf_rem_len = iov_size - user_buf_offset; + + let copied_len = cmp::min(packet_rem_payload.len(), user_buf_rem_len); + if copied_len == 0 { + // No more data to read from this packet + break; + } + payload_buffer.extend_from_slice(&packet_rem_payload[..copied_len]); + + if !packet.ancillary_taken { + name = name.or_else(|| packet.ancillary_data.name.take()); + ancillary_data_buffer.push_back(packet.ancillary_data.clone()); + packet.ancillary_taken = true; // Mark ancillary data as taken + } + + packet.read_offset += copied_len; + user_buf_offset += copied_len; + total_copied_len += copied_len; + if packet.read_offset >= packet.payload.len() { + // If the packet is fully read, remove it from the queue + self.packets.pop_front(); + } + } + + let mut msg_writer = MsgWriter::new(stream); + + msg_writer.write_name(name, name_buf_size, UdsStreamScheme::fpath_inner)?; + + let full_len = cmp::min(total_copied_len, iov_size); + msg_writer.write_payload(&payload_buffer, full_len, iov_size)?; + + let mut num_fds = 0; + for ancillary_data in ancillary_data_buffer.iter() { + num_fds += ancillary_data.num_fds; + } + if !msg_writer.write_rights(num_fds) { + log::warn!( + "serialize_to_msgstream: Buffer too small for SCM_RIGHTS, dropping {} FDs.", + num_fds + ); + self.drop_fds(num_fds)?; + } + + for option in options { + let result = match option { + SO_PASSCRED => { + let mut success = true; + for data in &ancillary_data_buffer { + if !msg_writer.write_credentials(&data.cred) { + success = false; + break; + } + } + success + } + _ => { + log::warn!( + "serialize_to_msgstream: Unsupported socket option for serialization: {}", + option + ); + return Err(Error::new(EOPNOTSUPP)); + } + }; + if !result { + log::warn!("serialize_to_msgstream: Buffer too small for ancillary data, stopping further serialization."); + break; + } + } + + Ok(msg_writer.len()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Copy)] +pub enum State { + Unbound, + Bound, + Listening, + Connecting, + Accepted, + Established, + Closed, +} + +impl Default for State { + fn default() -> Self { + Self::Unbound + } +} + +#[derive(Debug, Default)] +pub struct Socket { + primary_id: usize, + path: Option, + options: HashSet, + flags: usize, + state: State, + awaiting: VecDeque, + connection: Option, +} + +impl Socket { + fn new( + id: usize, + path: Option, + state: State, + options: HashSet, + flags: usize, + connection: Option, + ) -> Self { + Self { + primary_id: id, + path, + state, + options, + flags, + connection, + ..Default::default() + } + } + + fn accept(&mut self, primary_id: usize, awaiting_client_id: usize) -> Result { + if !self.is_listening() { + log::error!( + "accept(id: {}): Accept called on a non-listening socket.", + self.primary_id + ); + return Err(Error::new(EINVAL)); + } + Ok(Self::new( + primary_id, + self.path.clone(), + State::Established, + self.options.clone(), + self.flags, + Some(Connection::new(awaiting_client_id)), + )) + } + + fn establish(&mut self, peer: usize) -> Result<()> { + if self.state != State::Connecting { + log::error!( + "establish(id: {}): Cannot establish connection in state: {:?}", + self.primary_id, + self.state + ); + return Err(Error::new(EINVAL)); + } + self.state = State::Accepted; + self.connection = Some(Connection::new(peer)); + Ok(()) + } + + fn connect(&mut self, other: usize, flags: usize) -> Result<()> { + match self.state { + State::Unbound | State::Bound => { + // If the socket is unbound or bound, wait for the listener to start listening. + if flags & O_NONBLOCK == O_NONBLOCK { + self.awaiting.push_back(other); + Ok(()) + } else { + // If the connecting target is not a listening, + // the connecting socket will block until the socket + // is ready to accept. + Err(Error::new(EWOULDBLOCK)) + } + } + State::Listening => { + // If the socket is already listening, it can accept connections. + self.awaiting.push_back(other); + Ok(()) + } + _ => Err(Error::new(ECONNREFUSED)), + } + } + + // For socketpair, add the peer's id to the base socket's awaiting queue. + fn connect_socketpair(&mut self, peer: usize) { + self.awaiting.push_back(peer); + } + + fn is_listening(&self) -> bool { + self.state == State::Listening + } + + fn require_connection(&mut self) -> Result<&mut Connection> { + if let Some(connection) = &mut self.connection { + Ok(connection) + } else { + log::error!( + "Socket (id: {}): connection is None in require_connection", + self.primary_id + ); + Err(Error::new(EPROTO)) + } + } + + fn require_connected_connection(&mut self) -> Result<&mut Connection> { + match self.state { + State::Established | State::Accepted => self.require_connection(), + State::Closed => Err(Error::new(EPIPE)), + _ => Err(Error::new(ENOTCONN)), + } + } + + fn start_listening(&mut self) -> Result<()> { + if !matches!(self.state, State::Unbound | State::Bound) { + log::error!( + "start_listening(id: {}): Socket cannot listen in state {:?}.", + self.primary_id, + self.state + ); + return Err(Error::new(EINVAL)); + } + self.state = State::Listening; + Ok(()) + } + + fn serialize_to_msgstream( + &mut self, + stream: &mut [u8], + name_buf_size: usize, + iov_size: usize, + ) -> Result { + let options = self.options.clone(); + let connection = self.require_connected_connection()?; + connection.serialize_to_msgstream(stream, name_buf_size, iov_size, options) + } +} + +pub struct UdsStreamScheme<'sock> { + sockets: HashMap>>, + next_id: usize, + socket_paths: HashMap>>, + socket: &'sock SchemeSocket, +} + +impl<'sock> UdsStreamScheme<'sock> { + pub fn new(socket: &'sock SchemeSocket) -> Self { + Self { + sockets: HashMap::new(), + next_id: 0, + socket_paths: HashMap::new(), + socket, + } + } + + fn post_fevent(&self, id: usize, flags: usize) -> Result<()> { + let fevent_response = Response::post_fevent(id, flags); + match self + .socket + .write_response(fevent_response, SignalBehavior::Restart) + { + Ok(true) => Ok(()), // Write response success + Ok(false) => Err(Error::new(EAGAIN)), // Write response failed, retry. + Err(err) => Err(err), // Error writing response + } + } + + fn get_socket(&self, id: usize) -> Result<&Rc>, Error> { + self.sockets.get(&id).ok_or(Error::new(EBADF)) + } + + fn get_connected_peer(&self, id: usize) -> Result<(usize, Rc>), Error> { + let mut socket = self.get_socket(id)?.borrow_mut(); + + let remote_id = socket.require_connected_connection()?.peer; + let remote_rc = self.get_socket(remote_id).map_err(|e| { + log::error!("get_connected_peer(id: {}): Peer socket (id: {}) has vanished. Original error: {:?}", id, remote_id, e); + Error::new(EPIPE) + })?; + + if remote_rc.borrow().state == State::Closed { + log::error!( + "get_connected_peer(id: {}): Attempted to interact with a closed peer (id: {}).", + id, + remote_id + ); + return Err(Error::new(ECONNREFUSED)); + } + + Ok((remote_id, remote_rc.clone())) + } + + fn handle_unnamed_socket(&mut self, flags: usize) -> usize { + let new_id = self.next_id; + let new = Socket::new(new_id, None, State::Unbound, HashSet::new(), flags, None); + self.sockets.insert(new_id, Rc::new(RefCell::new(new))); + self.next_id += 1; + new_id + } + + fn call_inner( + &mut self, + id: usize, + payload: &mut [u8], + metadata: &[u64], + ctx: &CallerCtx, + ) -> Result { + let Some(verb) = SocketCall::try_from_raw(metadata[0] as usize) else { + log::error!("call_inner: Invalid verb in metadata: {:?}", metadata); + return Err(Error::new(EINVAL)); + }; + match verb { + SocketCall::Bind => self.handle_bind(id, &payload), + SocketCall::Connect => self.handle_connect(id, &payload), + SocketCall::SetSockOpt => self.handle_setsockopt(id, metadata[1] as i32, &payload), + SocketCall::GetSockOpt => self.handle_getsockopt(id, metadata[1] as i32, payload), + SocketCall::SendMsg => self.handle_sendmsg(id, payload, ctx), + SocketCall::RecvMsg => self.handle_recvmsg(id, payload), + _ => Err(Error::new(OPNOTSUP)), + } + } + + fn handle_bind(&mut self, id: usize, path_buf: &[u8]) -> Result { + let path = path_buf_to_string(path_buf)?; + + if self.socket_paths.contains_key(&path) { + log::error!("handle_bind: Path '{}' is already in use.", path); + return Err(Error::new(EADDRINUSE)); + } + + let socket_rc = self.get_socket(id)?.clone(); + let mut socket = socket_rc.borrow_mut(); + + if socket.state != State::Unbound { + log::error!( + "handle_bind(id: {}): Socket is already bound or connected (state: {:?})", + id, + socket.state + ); + return Err(Error::new(EINVAL)); + } + + socket.path = Some(path.clone()); + socket.state = State::Bound; + + self.socket_paths.insert(path, socket_rc.clone()); + + Ok(0) + } + + // There are three phases of connecting a socket: + // + // Phase 1: The listener is bound but not yet listening. + // The client is trying to connect. + // If the listener is not listening, the client will block + // and wait until the listener starts listening. + // + // Phase 2: The listener is now listening. + // The client is still trying to connect. + // The client pushes its ID to the listener's awaiting queue + // and sets its state to `Connecting`. + // The client then blocks, waiting for the listener to accept it. + // + // Phase 3: The listener accepts the client, changes its state to `Established`, + // and then changes the client's state to `Accepted`. + // The client detects that its state has changed to `Accepted` + // and changes its own state to `Established`. + // + // After these three phases, the socket connection is considered established. + // + // After these three phases, the socket connection is considered established. + // + // The reason why `connect` is complicated is that if the processing blocks, + // the SQE will be pushed back to the scheme's request queue, + // and the same SQE will be woken up later. + fn handle_connect(&mut self, id: usize, path_buf: &[u8]) -> Result { + let path = path_buf_to_string(path_buf)?; + let (listener_id, flags) = { + let listener_rc = self + .socket_paths + .get(&path) + .ok_or_else(|| Error::new(ECONNREFUSED))?; + + let client_rc = self.get_socket(id)?; + let mut client = client_rc.borrow_mut(); + + match client.state { + State::Connecting => { + // If the client is already Connecting + // Fence to prevent calling connect multiple times. + return Err(Error::new(EWOULDBLOCK)); + } + State::Established => { + return Err(Error::new(EISCONN)); + } + State::Accepted => { + // Phase 3: Socket is already connected + client.state = State::Established; + return Ok(0); + } + _ => {} + } + + // Phase 1: listener is bound but not yet listening + let mut listener = listener_rc.borrow_mut(); + let listener_id = listener.primary_id; + + listener.connect(id, client.flags)?; + // Phase 2: listener is now listening + client.state = State::Connecting; + + (listener_id, client.flags) + }; + // smoltcp sends writeable whenever a listener gets a + // client, we'll do the same too (but also readable, why + // not) + self.post_fevent(listener_id, (EVENT_READ | EVENT_WRITE).bits())?; + + // Blocking pattern + if flags & O_NONBLOCK == 0 { + return Err(Error::new(EWOULDBLOCK)); + } + + // Non-blocking pattern + Ok(0) + } + + fn handle_setsockopt(&mut self, id: usize, option: i32, value_slice: &[u8]) -> Result { + let socket_rc = self.get_socket(id)?; + let mut socket = socket_rc.borrow_mut(); + + match option { + SO_PASSCRED => { + let value = read_num::(value_slice)?; + if value != 0 { + socket.options.insert(SO_PASSCRED); + } else { + socket.options.remove(&SO_PASSCRED); + } + Ok(value_slice.len()) + } + _ => { + log::warn!( + "socket_setsockopt(id: {}): Unsupported option: {}", + id, + option + ); + Err(Error::new(ENOPROTOOPT)) + } + } + } + + fn handle_getsockopt(&mut self, id: usize, option: i32, payload: &mut [u8]) -> Result { + match option { + SO_DOMAIN => { + payload.fill(0); + if payload.len() < mem::size_of::() { + log::error!( + "socket_getsockopt(id: {}): SO_DOMAIN payload buffer is too small. len: {}", + id, + payload.len() + ); + return Err(Error::new(ENOBUFS)); + } + let domain = AF_UNIX.to_le_bytes(); + payload[..domain.len()].copy_from_slice(&domain); + Ok(domain.len()) + } + _ => { + log::warn!( + "socket_getsockopt(id: {}): Unsupported option: {}", + id, + option + ); + Err(Error::new(ENOPROTOOPT)) + } + } + } + + fn handle_sendmsg(&mut self, id: usize, msg_stream: &[u8], ctx: &CallerCtx) -> Result { + if msg_stream.is_empty() { + log::error!("msg_stream is empty, returning EINVAL."); + return Err(Error::new(EINVAL)); + } + + let name = self.get_socket(id)?.borrow().path.clone(); + let (remote_id, remote_rc) = self.get_connected_peer(id)?; + + let bytes_written = + Self::sendmsg_inner(&mut remote_rc.borrow_mut(), name, msg_stream, ctx)?; + self.post_fevent(remote_id, EVENT_READ.bits())?; + Ok(bytes_written) + } + + fn sendmsg_inner( + socket: &mut Socket, + name: Option, + msg_stream: &[u8], + ctx: &CallerCtx, + ) -> Result { + if msg_stream.is_empty() { + log::error!("sendmsg_inner: msg_stream is empty."); + return Err(Error::new(EINVAL)); + } + + let connection = socket.require_connected_connection()?; + let packet = DataPacket::from_stream(msg_stream, name, ctx)?; + let payload_len = packet.len(); + connection.packets.push_back(packet); + + Ok(payload_len) + } + + fn handle_recvmsg(&mut self, id: usize, msg_stream: &mut [u8]) -> Result { + let socket_rc = self.get_socket(id)?; + let mut socket = socket_rc.borrow_mut(); + let flags = socket.flags; + let connection = match &mut socket.state { + State::Established | State::Accepted => socket.require_connection()?, + State::Closed => { + // Remote dropped, send EOF + return Self::write_eof(msg_stream); + } + State::Listening => { + log::warn!("socket_recvmsg: Called on a listening socket, returning EOPNOTSUPP."); + return Err(Error::new(EOPNOTSUPP)); + } + _ => return Err(Error::new(ENOTCONN)), + }; + + if connection.packets.is_empty() { + return if connection.is_peer_shutdown { + // EOF, no data to read + return Self::write_eof(msg_stream); + } else if (flags as usize) & O_NONBLOCK == O_NONBLOCK { + Err(Error::new(EAGAIN)) + } else { + Err(Error::new(EWOULDBLOCK)) + }; + } + Self::recvmsg_inner(&mut socket, msg_stream) + } + + fn write_eof(buffer: &mut [u8]) -> Result { + // Write EOF to the buffer + let target = buffer.get_mut(..MIN_RECV_MSG_LEN).ok_or_else(|| { + log::error!("write_eof: Buffer is too small to write EOF, returning EINVAL."); + Error::new(EINVAL) + })?; + target.fill(0); // Fill the buffer with zeros to indicate EOF + Ok(MIN_RECV_MSG_LEN) + } + + fn recvmsg_inner(socket: &mut Socket, msg_stream: &mut [u8]) -> Result { + let (prepared_name_len, prepared_whole_iov_size, _) = read_msghdr_info(msg_stream)?; + + let written_len = socket.serialize_to_msgstream( + msg_stream, + prepared_name_len, + prepared_whole_iov_size, + )?; + + Ok(written_len) + } + + fn accept_connection( + &mut self, + listener_socket: &mut Socket, + client_id: usize, + ) -> Result> { + let (new_id, new) = { + let Ok(client_rc) = self.get_socket(client_id) else { + return Ok(None); // Client socket has been closed, nothing to accept + }; + let new_id = self.next_id; + let new = listener_socket.accept(new_id, client_id)?; + + let mut client_socket = client_rc.borrow_mut(); + client_socket.establish(new_id)?; + (new_id, new) + }; + + self.next_id += 1; + self.sockets.insert(new_id, Rc::new(RefCell::new(new))); + self.post_fevent(client_id, (EVENT_READ | EVENT_WRITE).bits())?; + Ok(Some(OpenResult::ThisScheme { + number: new_id, + flags: NewFdFlags::empty(), + })) + } + + fn handle_accept(&mut self, id: usize, socket: &mut Socket) -> Result> { + let flags = socket.flags; + if !socket.is_listening() { + log::error!( + "socket_accept: Socket state is not Listening for id: {}", + id + ); + return Err(Error::new(EINVAL)); + } + // Try to accept a waiting connection + let Some(client_id) = socket.awaiting.pop_front() else { + if flags & O_NONBLOCK == O_NONBLOCK { + return Ok(Some(OpenResult::WouldBlock)); + } else { + return Err(Error::new(EWOULDBLOCK)); + } + }; + Ok(self.accept_connection(socket, client_id)?) + } + + // Transition a Bound or Unbound socket to the Listening state. + fn handle_start_listening(&mut self, socket_rc: &Rc>) -> Result<()> { + let path = { + let mut socket = socket_rc.borrow_mut(); + socket.start_listening()?; + socket.path.clone() + }; + + if let Some(path) = path { + if let Some(existing_socket_rc) = self.socket_paths.get(&path) { + if !Rc::ptr_eq(socket_rc, existing_socket_rc) { + log::error!("handle_start_listening: Path '{}' is already in use.", path); + return Err(Error::new(EADDRINUSE)); + } + } + self.socket_paths.insert(path, socket_rc.clone()); + } + Ok(()) + } + + // Handle a `dup` call for `b"listen"`. + // If the socket is not yet listening, it transitions it to the Listening state. + // If it is already listening, it tries to accept a pending connection. + fn handle_listen(&mut self, id: usize) -> Result { + loop { + let socket_rc = self.get_socket(id)?.clone(); + let is_listening = socket_rc.borrow().is_listening(); + + if is_listening { + let mut socket = socket_rc.borrow_mut(); + match self.handle_accept(id, &mut socket)? { + Some(result) => return Ok(result), + None => continue, + } + } else { + self.handle_start_listening(&socket_rc)?; + continue; + } + } + } + + fn handle_connect_socketpair(&mut self, id: usize) -> Result { + let new_id = self.next_id; + let flags = self.get_socket(id)?.borrow().flags; + let new = Socket::new(new_id, None, State::Connecting, HashSet::new(), flags, None); + { + let socket_rc = self.get_socket(id)?; + let mut socket = socket_rc.borrow_mut(); + + if socket.state == State::Closed { + log::error!( + "socket_connect_socketpair: Base socket {} is already closed.", + id + ); + return Err(Error::new(EPIPE)); + } + socket.connect_socketpair(new_id); + } + + // smoltcp sends writeable whenever a listener gets a + // client, we'll do the same too (but also readable, + // why not) + self.post_fevent(id, (EVENT_READ | EVENT_WRITE).bits())?; + + self.sockets.insert(new_id, Rc::new(RefCell::new(new))); + + self.next_id += 1; + + Ok(OpenResult::ThisScheme { + number: new_id, + flags: NewFdFlags::empty(), + }) + } + + fn handle_recvfd(&mut self, id: usize) -> Result { + let socket_rc = self.get_socket(id)?; + let mut socket = socket_rc.borrow_mut(); + + match socket.state { + State::Established | State::Accepted => { + let connection = socket.require_connected_connection()?; + let fd = connection.fds.pop_front().ok_or(Error::new(EWOULDBLOCK))?; + Ok(OpenResult::OtherScheme { fd }) + } + State::Closed => Err(Error::new(EPIPE)), + State::Listening => Err(Error::new(EOPNOTSUPP)), + _ => Err(Error::new(ENOTCONN)), + } + } + + fn write_inner(&mut self, receiver_id: usize, buf: &[u8], ctx: &CallerCtx) -> Result { + let receiver_rc = self.get_socket(receiver_id)?; + let mut receiver = receiver_rc.borrow_mut(); + let name = receiver.path.clone(); + + let connection = receiver.require_connected_connection()?; + + if !buf.is_empty() { + // Send readable only if it wasn't readable before + let ancillary_data = AncillaryData::new( + Credential::new(ctx.pid as i32, ctx.uid as i32, ctx.gid as i32), + name, + ); + let packet = DataPacket::new(buf.to_vec(), ancillary_data); + connection.packets.push_back(packet); + self.post_fevent(receiver_id, EVENT_READ.bits())?; + } + + Ok(buf.len()) + } + + fn sendfd_inner( + &mut self, + receiver_id: usize, + sendfd_request: &SendFdRequest, + ) -> Result { + let mut new_fd = usize::MAX; + if let Err(e) = + sendfd_request.obtain_fd(&self.socket, FobtainFdFlags::empty(), Err(&mut new_fd)) + { + log::error!("sendfd_inner: obtain_fd failed with error: {:?}", e); + return Err(e); + } + let receiver_rc = self.get_socket(receiver_id)?; + let mut receiver = receiver_rc.borrow_mut(); + + let connection = receiver.require_connected_connection()?; + connection.fds.push_back(new_fd); + + self.post_fevent(receiver_id, EVENT_READ.bits())?; + + Ok(new_fd) + } + + fn read_inner(connection: &mut Connection, buf: &mut [u8], flags: u32) -> Result { + let mut total_copied_len = 0; + let mut user_buf_offset = 0; + + while user_buf_offset < buf.len() { + let Some(packet) = connection.packets.front_mut() else { + // No more packets to read + break; + }; + + let packet_rem_payload = &packet.payload[packet.read_offset..]; + + let user_buf_rem_len = buf.len() - user_buf_offset; + + let copied_len = cmp::min(packet_rem_payload.len(), user_buf_rem_len); + if copied_len == 0 { + // No more data to read from this packet + break; + } + buf[user_buf_offset..user_buf_offset + copied_len] + .copy_from_slice(&packet_rem_payload[..copied_len]); + + if packet.read_offset == 0 { + packet.ancillary_taken = true; // Mark ancillary data as taken + } + + packet.read_offset += copied_len; + user_buf_offset += copied_len; + total_copied_len += copied_len; + if packet.read_offset >= packet.payload.len() { + // If the packet is fully read, remove it from the queue + connection.packets.pop_front(); + } + } + + if total_copied_len > 0 { + Ok(total_copied_len) + } else if connection.is_peer_shutdown { + Ok(0) // EOF, no data to read + } else if (flags as usize) & O_NONBLOCK == O_NONBLOCK { + Err(Error::new(EAGAIN)) + } else { + Err(Error::new(EWOULDBLOCK)) + } + } + + fn handle_listening_closure(&mut self, socket_rc: Rc>) { + let socket = socket_rc.borrow(); + if let Some(path) = &socket.path { + self.socket_paths.remove(path); + } + + // Notify all waiting clients about listener closure + for client_id in &socket.awaiting { + if let Ok(client_rc) = self.get_socket(*client_id) { + let mut client = client_rc.borrow_mut(); + client.state = State::Closed; + let _ = self.post_fevent(*client_id, EVENT_READ.bits()); + + drop(client); + } + } + } + + fn handle_other_closure(&mut self, socket_rc: Rc>) { + // If this is the last reference to the socket, it's safe to remove the socket path. + if matches!( + socket_rc.borrow().state, + State::Established | State::Accepted + ) { + let mut socket = socket_rc.borrow_mut(); + let Ok(connection) = socket.require_connection() else { + return; + }; + let Ok(remote_rc) = self.get_socket(connection.peer) else { + return; + }; + let mut remote = remote_rc.borrow_mut(); + let Ok(connection) = remote.require_connection() else { + return; + }; + connection.is_peer_shutdown = true; + let _ = self.post_fevent(remote.primary_id, EVENT_READ.bits()); + } + + if Rc::strong_count(&socket_rc) == 2 { + if let Some(path) = socket_rc.borrow().path.clone() { + // If this is the last reference to the socket, remove the path from the registry + self.socket_paths.remove(&path); + } + } + socket_rc.borrow_mut().state = State::Closed; + } + + fn fpath_inner(path: &String, buf: &mut [u8]) -> Result { + // Write scheme name + const PREFIX: &[u8] = b"/scheme/uds_stream/"; + let len = cmp::min(PREFIX.len(), buf.len()); + buf[..len].copy_from_slice(&PREFIX[..len]); + if len < PREFIX.len() { + return Ok(len); + } + + // Write path + let len = cmp::min(path.len(), buf.len() - PREFIX.len()); + buf[PREFIX.len()..][..len].copy_from_slice(&path.as_bytes()[..len]); + + Ok(PREFIX.len() + len) + } +} + +impl<'sock> SchemeSync for UdsStreamScheme<'sock> { + fn open(&mut self, path: &str, flags: usize, _ctx: &CallerCtx) -> Result { + let new_id = if path.is_empty() { + self.handle_unnamed_socket(flags) + } else { + log::error!( + "open(path: '{}'): Attempting to open a named socket, which is not supported.", + path + ); + return Err(Error::new(EINVAL)); + }; + Ok(OpenResult::ThisScheme { + number: new_id, + flags: NewFdFlags::empty(), + }) + } + + fn call( + &mut self, + id: usize, + payload: &mut [u8], + metadata: &[u64], + ctx: &CallerCtx, + ) -> Result { + self.call_inner(id, payload, metadata, ctx) + } + + fn dup(&mut self, id: usize, buf: &[u8], _ctx: &CallerCtx) -> Result { + match buf { + b"listen" => self.handle_listen(id), + b"connect" => self.handle_connect_socketpair(id), + b"recvfd" => self.handle_recvfd(id), + _ => Err(Error::new(EINVAL)), + } + } + + fn write( + &mut self, + id: usize, + buf: &[u8], + _offset: u64, + _flags: u32, + ctx: &CallerCtx, + ) -> Result { + let (receiver_id, _) = self.get_connected_peer(id)?; + self.write_inner(receiver_id, buf, ctx) + } + + fn fpath(&mut self, id: usize, buf: &mut [u8], _ctx: &CallerCtx) -> Result { + let socket_rc = self.get_socket(id)?; + let socket = socket_rc.borrow(); + + let path = socket.path.as_ref().ok_or(Error::new(EBADF))?; + Ok(Self::fpath_inner(path, buf)?) + } + + fn fsync(&mut self, id: usize, _ctx: &CallerCtx) -> Result<()> { + self.get_socket(id).and(Ok(())) + } + + fn read( + &mut self, + id: usize, + buf: &mut [u8], + _offset: u64, + flags: u32, + _ctx: &CallerCtx, + ) -> Result { + let socket_rc = self.get_socket(id)?; + let mut socket = socket_rc.borrow_mut(); + + match socket.state { + State::Established | State::Accepted => { + let connection = socket.require_connected_connection()?; + Self::read_inner(connection, buf, flags) + } + State::Closed => Ok(0), + State::Listening => Err(Error::new(EOPNOTSUPP)), + _ => Err(Error::new(ENOTCONN)), + } + } + + fn on_sendfd(&mut self, sendfd_request: &SendFdRequest) -> Result { + let id = sendfd_request.id(); + let (receiver_id, _) = self.get_connected_peer(id)?; + + self.sendfd_inner(receiver_id, sendfd_request) + } + + fn on_close(&mut self, id: usize) { + let Some(socket_rc) = self.sockets.remove(&id) else { + return; + }; + + let state = socket_rc.borrow().state; + match state { + State::Listening => { + self.handle_listening_closure(socket_rc); + } + _ => { + self.handle_other_closure(socket_rc); + } + } + } + + fn fcntl(&mut self, id: usize, cmd: usize, arg: usize, _ctx: &CallerCtx) -> Result { + let socket_rc = self.get_socket(id)?; + let mut socket = socket_rc.borrow_mut(); + match cmd { + F_GETFL => Ok(socket.flags), + F_SETFL => { + socket.flags = arg; + Ok(0) + } + _ => Err(Error::new(EINVAL)), + } + } +}