Merge branch 'scheme0.6-uds' into 'main'

feat: Implement uds_dgram and uds_stream schemes.

See merge request redox-os/base!16

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"

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",

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<usize> {
+        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::<ProcMeta>())
+            .and_then(|b| plain::from_mut_bytes(b).ok())
+            .ok_or(Error::new(EBADF))? = metadata;
+        Ok(size_of::<ProcMeta>())
+    }
     fn on_read(&mut self, id: usize, offset: u64, buf: &mut [u8]) -> Result<usize> {
         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::<ProcMeta>())
-                    .and_then(|b| plain::from_mut_bytes(b).ok())
-                    .ok_or(Error::new(EINVAL))? = metadata;
-                Ok(size_of::<ProcMeta>())
-            }
+            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<u32> {
@@ -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),

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"

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<bool>, 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<bool>, 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(())

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<usize> {
+        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<String>,
+    state: State,
+    peer: Option<usize>,
+    messages: VecDeque<DataPacket>,
+    options: HashSet<i32>,
+    fds: VecDeque<usize>,
+    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<usize, Rc<RefCell<Socket>>>,
+    next_id: usize,
+    socket_paths: HashMap<String, Rc<RefCell<Socket>>>,
+    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<RefCell<Socket>>, Error> {
+        self.sockets.get(&id).ok_or(Error::new(EBADF))
+    }
+
+    fn get_connected_peer(&self, id: usize) -> Result<(usize, Rc<RefCell<Socket>>), 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<usize> {
+        // metadata to Vec<u8>
+        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<usize> {
+        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<usize> {
+        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<usize> {
+        let socket_rc = self.get_socket(id)?;
+        let mut socket = socket_rc.borrow_mut();
+
+        match option {
+            SO_PASSCRED => {
+                let value = read_num::<i32>(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<usize> {
+        match option {
+            SO_DOMAIN => {
+                payload.fill(0);
+                if payload.len() < mem::size_of::<i32>() {
+                    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<usize> {
+        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<String>,
+        msg_stream: &[u8],
+        ctx: &CallerCtx,
+    ) -> Result<usize> {
+        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<usize> {
+        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<usize> {
+        // 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<OpenResult> {
+        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<OpenResult> {
+        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<OpenResult> {
+        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<usize> {
+        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<usize> {
+        // 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<usize> {
+        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<usize> {
+        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<OpenResult> {
+        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<usize> {
+        self.call_inner(id, payload, metadata, ctx)
+    }
+
+    fn dup(&mut self, id: usize, buf: &[u8], _ctx: &CallerCtx) -> Result<OpenResult> {
+        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<usize> {
+        self.write_inner(id, buf, ctx)
+    }
+
+    fn fpath(&mut self, id: usize, buf: &mut [u8], _ctx: &CallerCtx) -> Result<usize> {
+        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<usize> {
+        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<usize> {
+        self.sendfd_inner(sendfd_request)
+    }
+
+    fn fcntl(&mut self, id: usize, cmd: usize, arg: usize, _ctx: &CallerCtx) -> Result<usize> {
+        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))
+            }
+        }
+    }
+}

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::<usize>() * 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::<i32>();
+const CMSG_TYPE_SIZE: usize = mem::size_of::<i32>();
+const CMSG_DATA_LEN_SIZE: usize = mem::size_of::<usize>();
+const PID_SIZE: usize = mem::size_of::<i32>();
+const UID_SIZE: usize = mem::size_of::<i32>();
+const GID_SIZE: usize = mem::size_of::<i32>();
+
+#[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<String>,
+}
+impl AncillaryData {
+    fn new(cred: Credential, name: Option<String>) -> Self {
+        Self {
+            cred,
+            name,
+            ..Default::default()
+        }
+    }
+}
+
+struct AncillaryDataHeader {
+    level: i32,
+    c_type: i32,
+    data_len: usize,
+}
+impl AncillaryDataHeader {
+    fn from_stream(stream: &[u8]) -> Result<Option<Self>> {
+        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::<i32>(&stream[cursor..])?;
+        cursor += mem::size_of::<i32>();
+
+        let cmsg_type = read_num::<i32>(&stream[cursor..])?;
+        cursor += mem::size_of::<i32>();
+
+        let cmsg_data_len = read_num::<usize>(&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<u8>,
+    ancillary_data: AncillaryData,
+    // Only for stream packets
+    read_offset: usize,
+    ancillary_taken: bool,
+}
+
+impl DataPacket {
+    fn new(payload: Vec<u8>, 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<String>, ctx: &CallerCtx) -> Result<Self> {
+        let mut cursor: usize = 0;
+        let payload_len = read_num::<usize>(&stream[cursor..])?;
+        cursor += mem::size_of::<usize>();
+        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::<usize>(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<T>(buffer: &[u8]) -> Result<T, Error>
+where
+    T: NumFromBytes,
+{
+    T::from_le_bytes_slice(buffer)
+}
+trait NumFromBytes: Sized {
+    fn from_le_bytes_slice(buffer: &[u8]) -> Result<Self, Error>;
+}
+impl NumFromBytes for i32 {
+    fn from_le_bytes_slice(buffer: &[u8]) -> Result<Self> {
+        Ok(i32::from_le_bytes(
+            buffer
+                .get(..mem::size_of::<i32>())
+                .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<Self> {
+        Ok(u32::from_le_bytes(
+            buffer
+                .get(..mem::size_of::<u32>())
+                .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<Self> {
+        Ok(usize::from_le_bytes(
+            buffer
+                .get(..mem::size_of::<usize>())
+                .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::<ProcMeta>()];
+    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::<u32>(&buffer[cursor..])?;
+    cursor += mem::size_of::<u32>() * 3;
+    let uid = read_num::<u32>(&buffer[cursor..])?;
+    cursor += mem::size_of::<u32>() * 3;
+    let gid = read_num::<u32>(&buffer[cursor..])?;
+    Ok((pid, uid, gid))
+}
+
+fn read_msghdr_info(stream: &mut [u8]) -> Result<(usize, usize, usize)> {
+    if stream.len() < mem::size_of::<usize>() * 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::<usize>(&stream[cursor..])?;
+    cursor += mem::size_of::<usize>();
+    let prepared_whole_iov_size = read_num::<usize>(&stream[cursor..])?;
+    cursor += mem::size_of::<usize>();
+    let prepared_msg_controllen = read_num::<usize>(&stream[cursor..])?;
+    cursor += mem::size_of::<usize>();
+    // Clear the stream buffer
+    stream[..cursor].copy_from_slice(&[0u8; mem::size_of::<usize>() * 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<String>,
+        name_buf_size: usize,
+        name_write_fn: fn(&String, &mut [u8]) -> Result<usize>,
+    ) -> Result<()> {
+        if self.buffer.len() < self.written_len + mem::size_of::<usize>() {
+            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::<usize>(),
+            );
+            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::<usize>()
+                    ..self.written_len + mem::size_of::<usize>() + copy_len],
+            )?;
+            self.buffer[self.written_len..self.written_len + mem::size_of::<usize>()]
+                .copy_from_slice(&name_len.to_le_bytes());
+            self.written_len += mem::size_of::<usize>() + name_len;
+            Ok(())
+        } else {
+            self.written_len += mem::size_of::<usize>();
+            Ok(())
+        }
+    }
+
+    fn write_payload(&mut self, payload: &[u8], full_len: usize, iov_size: usize) -> Result<usize> {
+        let Some(payload_len_buffer) = self
+            .buffer
+            .get_mut(self.written_len..self.written_len + mem::size_of::<usize>())
+        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::<usize>();
+
+        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<String> {
+    match std::str::from_utf8(path_buf) {
+        Ok("") => Err(Error::new(EINVAL)),
+        Ok(s) => Ok(s.to_string()),
+        Err(_) => Err(Error::new(EINVAL)),
+    }
+}