vasilito/RedBear-OS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'integrate-bind-connect-redoxfs' into 'main'

Browse Source 
feat: Reimplement bind and connect operation using token and switch GetProcCredentials to capability-based approach.

See merge request redox-os/base!19
This commit is contained in:
Jeremy Soller
2025-07-21 07:56:03 -06:00
parent 257e07209d 9a8af7c993
commit d03f2cddd1
8 changed files with 392 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+3 -3
View File
@@ -451,7 +451,7 @@ dependencies = [
"anyhow",
"libc",
"libredox",
"log",
"rand",
"redox-daemon",
"redox-rt",
"redox-scheme 0.7.0",
@@ -837,9 +837,9 @@ dependencies = [
[[package]]
name = "redox_syscall"
version = "0.5.13"
version = "0.5.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0d04b7d0ee6b4a0207a0a7adb104d23ecb0b47d6beae7152d0fa34b692b29fd6"
checksum = "de3a5d9f0aba1dbcec1cc47f0ff94a4b778fe55bca98a6dfa92e4e094e57b1c4"
dependencies = [
"bitflags 2.9.0",
]
bootstrap/Cargo.lock
Generated
+2 -2
View File
@@ -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#518798f64e313a2d9067a52b6a3d5970c8208ae5"
source = "git+https://gitlab.redox-os.org/redox-os/relibc.git#648d3be78ef1b16da46a14bc83ef998d45fa555a"
[[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#518798f64e313a2d9067a52b6a3d5970c8208ae5"
source = "git+https://gitlab.redox-os.org/redox-os/relibc.git#648d3be78ef1b16da46a14bc83ef998d45fa555a"
dependencies = [
"bitflags",
"generic-rt",
bootstrap/src/procmgr.rs
+35 -22
View File
@@ -34,7 +34,8 @@ use syscall::{
sig_bit, ContextStatus, ContextVerb, CtxtStsBuf, Error, Event, EventFlags, FobtainFdFlags,
MapFlags, ProcSchemeAttrs, Result, SenderInfo, SetSighandlerData, SigProcControl, Sigcontrol,
EACCES, EAGAIN, EBADF, EBADFD, ECANCELED, ECHILD, EEXIST, EINTR, EINVAL, ENOENT, ENOSYS,
EOPNOTSUPP, EOWNERDEAD, EPERM, ERESTART, ESRCH, EWOULDBLOCK, O_CREAT, PAGE_SIZE,
EOPNOTSUPP, EOWNERDEAD, EPERM, ERESTART, ESRCH, EWOULDBLOCK, O_ACCMODE, O_CREAT, O_RDONLY,
PAGE_SIZE,
};
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
@@ -511,6 +512,9 @@ enum Handle {
// TODO: stateless API, perhaps using intermediate daemon for providing a file-like API
Ps(Vec<u8>),
// A handle that grants the holder the capability to obtain process credentials.
ProcCredsCapability,
}
#[derive(Clone, Copy, Debug)]
@@ -790,6 +794,18 @@ impl<'a> ProcScheme<'a> {
flags: NewFdFlags::POSITIONED,
}
}
"proc-creds-capability" => {
if ctx.uid != 0 {
return Err(Error::new(EACCES));
}
if flags & O_ACCMODE != O_RDONLY {
return Err(Error::new(EINVAL));
}
OpenResult::ThisScheme {
number: self.handles.insert(Handle::ProcCredsCapability),
flags: NewFdFlags::empty(),
}
}
_ => return Err(Error::new(ENOENT)),
})
@@ -827,7 +843,9 @@ impl<'a> ProcScheme<'a> {
buf[..len].copy_from_slice(&src_buf[..len]);
Ok(len)
}
Handle::Init | Handle::Thread(_) => return Err(Error::new(EBADF)),
Handle::Init | Handle::Thread(_) | Handle::ProcCredsCapability => {
return Err(Error::new(EBADF))
}
}
}
fn on_dup(&mut self, old_id: usize, buf: &[u8]) -> Result<OpenResult> {
@@ -874,7 +892,7 @@ impl<'a> ProcScheme<'a> {
fd: syscall::dup(*thread.fd, buf)?,
})
}
Handle::Init | Handle::Ps(_) => Err(Error::new(EBADF)),
Handle::Init | Handle::Ps(_) | Handle::ProcCredsCapability => Err(Error::new(EBADF)),
}
}
fn on_call(
@@ -1042,27 +1060,23 @@ 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,
))
}
ProcCall::GetProcCredentials => Response::ready_err(EACCES, op),
}
}
Handle::Ps(_) => Response::ready_err(EOPNOTSUPP, op),
Handle::ProcCredsCapability => {
let Some(verb) = ProcCall::try_from_raw(metadata[0] as usize) else {
log::info!("Invalid proc call: {metadata:?}");
return Response::ready_err(EINVAL, op);
};
match verb {
ProcCall::GetProcCredentials => Ready(Response::new(
self.read_process_metadata(ProcessId(metadata[1] as usize), payload),
op,
)),
_ => Response::ready_err(EINVAL, op),
}
}
}
}
fn on_getpgid(&mut self, caller_pid: ProcessId, target_pid: ProcessId) -> Result<ProcessId> {
@@ -2520,7 +2534,6 @@ impl<'a> ProcScheme<'a> {
log::trace!("{} thread_lookup", self.thread_lookup.len());
log::trace!("{} next_id", self.next_internal_id);
Ok(string.into_bytes())
}
}
ipcd/Cargo.toml
+1 -1
View File
@@ -12,4 +12,4 @@ redox-rt = { git = "https://gitlab.redox-os.org/redox-os/relibc.git", default-fe
redox_event = "0.4.0"
anyhow = "1.0.97"
libc = "0.2"
log = "0.4"
rand = { version = "0.8", features = ["small_rng"] }
ipcd/src/main.rs
+12 -8
View File
@@ -52,16 +52,20 @@ fn inner(daemon: redox_daemon::Daemon) -> anyhow::Result<()> {
// 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 uds_stream = Mutex::new(
UdsStreamScheme::new(&uds_stream_socket)
.map_err(|e| anyhow::anyhow!("failed to create uds stream scheme: {e}"))?,
);
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 uds_dgram = Mutex::new(
UdsDgramScheme::new(&uds_dgram_socket)
.map_err(|e| anyhow::anyhow!("failed to create uds dgram scheme: {e}"))?,
);
let mut uds_dgram_handler = ReadinessBased::new(&uds_dgram_socket, 16);
let uds_dgram_mutex = Mutex::new(uds_dgram);
daemon.ready().unwrap();
@@ -189,11 +193,11 @@ fn inner(daemon: redox_daemon::Daemon) -> anyhow::Result<()> {
}
// 2. Process requests
uds_stream_handler.process_requests(|| uds_stream_mutex.lock().unwrap());
uds_stream_handler.process_requests(|| uds_stream.lock().unwrap());
// 3.Poll all blocking requests
uds_stream_handler
.poll_all_requests(|| uds_stream_mutex.lock().unwrap())
.poll_all_requests(|| uds_stream.lock().unwrap())
.map_err(|e| anyhow::anyhow!("error occured in poll_all_requests: {e}"))?;
// 3. Write responses
@@ -222,11 +226,11 @@ fn inner(daemon: redox_daemon::Daemon) -> anyhow::Result<()> {
}
// 2. Process requests
uds_dgram_handler.process_requests(|| uds_dgram_mutex.lock().unwrap());
uds_dgram_handler.process_requests(|| uds_dgram.lock().unwrap());
// 3.Poll all blocking requests
uds_dgram_handler
.poll_all_requests(|| uds_dgram_mutex.lock().unwrap())
.poll_all_requests(|| uds_dgram.lock().unwrap())
.map_err(|e| anyhow::anyhow!("error occured in poll_all_requests: {e}"))?;
// 3. Write responses
ipcd/src/uds/dgram.rs
+146 -58
View File
@@ -1,11 +1,13 @@
//! 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,
get_uid_gid_from_pid, path_buf_to_str, read_msghdr_info, read_num, AncillaryData, Credential,
DataPacket, MsgWriter, MAX_DGRAM_MSG_LEN,
};
use libc::{AF_UNIX, SO_DOMAIN, SO_PASSCRED};
use rand::rngs::SmallRng;
use rand::{RngCore, SeedableRng};
use redox_rt::protocol::SocketCall;
use redox_scheme::{
scheme::SchemeSync, CallerCtx, OpenResult, Response, SendFdRequest, SignalBehavior,
@@ -18,7 +20,7 @@ use std::{
mem,
rc::Rc,
};
use syscall::{error::*, flag::*, schemev2::NewFdFlags, Error, FobtainFdFlags};
use syscall::{error::*, flag::*, schemev2::NewFdFlags, Error, FobtainFdFlags, Stat};
impl DataPacket {
pub fn serialize_to_stream(
@@ -40,7 +42,7 @@ impl DataPacket {
// 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!(
eprintln!(
"serialize_to_stream: Buffer too small for SCM_RIGHTS, dropping {} FDs.",
self.ancillary_data.num_fds
);
@@ -51,7 +53,7 @@ impl DataPacket {
let result = match *option {
SO_PASSCRED => msg_writer.write_credentials(&self.ancillary_data.cred),
_ => {
log::warn!(
eprintln!(
"serialize_to_stream: Unsupported socket option for serialization: {}",
option
);
@@ -59,7 +61,7 @@ impl DataPacket {
}
};
if !result {
log::warn!("serialize_to_stream: Buffer too small for ancillary data, stopping further serialization.");
eprintln!("serialize_to_stream: Buffer too small for ancillary data, stopping further serialization.");
break;
}
}
@@ -91,13 +93,14 @@ pub struct Socket {
options: HashSet<i32>,
fds: VecDeque<usize>,
flags: usize,
issued_token: Option<u64>,
}
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);
eprintln!("Socket::drop_fds: Attempted to drop FD #{} of {}, but fd queue is empty. State inconsistency.", i + 1, num_fd);
return Err(Error::new(EINVAL));
}
}
@@ -109,17 +112,26 @@ pub struct UdsDgramScheme<'sock> {
sockets: HashMap<usize, Rc<RefCell<Socket>>>,
next_id: usize,
socket_paths: HashMap<String, Rc<RefCell<Socket>>>,
socket_tokens: HashMap<u64, Rc<RefCell<Socket>>>,
socket: &'sock SchemeSocket,
proc_creds_capability: usize,
rng: SmallRng,
}
impl<'sock> UdsDgramScheme<'sock> {
pub fn new(socket: &'sock SchemeSocket) -> Self {
Self {
pub fn new(socket: &'sock SchemeSocket) -> Result<Self> {
Ok(Self {
sockets: HashMap::new(),
next_id: 0,
socket_paths: HashMap::new(),
socket_tokens: HashMap::new(),
socket,
}
proc_creds_capability: syscall::open(
"/scheme/proc/proc-creds-capability",
syscall::O_RDONLY,
)?,
rng: SmallRng::from_entropy(),
})
}
fn post_fevent(&self, id: usize, flags: usize) -> Result<()> {
@@ -144,15 +156,14 @@ impl<'sock> UdsDgramScheme<'sock> {
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);
eprintln!("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!(
eprintln!(
"get_connected_peer(id: {}): Attempted to interact with a closed peer (id: {}).",
id,
remote_id
id, remote_id
);
return Err(Error::new(ECONNREFUSED));
}
@@ -180,7 +191,7 @@ impl<'sock> UdsDgramScheme<'sock> {
) -> 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);
eprintln!("call_inner: Invalid verb in metadata: {:?}", metadata);
return Err(Error::new(EINVAL));
};
match verb {
@@ -190,54 +201,71 @@ impl<'sock> UdsDgramScheme<'sock> {
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),
SocketCall::Unbind => self.handle_unbind(id),
SocketCall::GetToken => self.handle_get_token(id, payload),
_ => Err(Error::new(EOPNOTSUPP)),
}
}
fn handle_bind(&mut self, id: usize, path_buf: &[u8]) -> Result<usize> {
let path = path_buf_to_string(path_buf)?;
let path = path_buf_to_str(path_buf)?;
// Check if path is already bound to a server
if self.socket_paths.contains_key(&path) {
log::warn!(
if self.socket_paths.contains_key(path) {
eprintln!(
"handle_bind(id: {}): Address '{}' already in use.",
id,
path
id, path
);
return Err(Error::new(EADDRINUSE));
}
let socket_rc = self.get_socket(id)?.clone();
let mut socket = socket_rc.borrow_mut();
let path_owned: String;
let token: u64;
{
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));
if socket.state != State::Unbound {
eprintln!(
"handle_bind(id: {}): Socket is already bound or connected (state: {:?})",
id, socket.state
);
return Err(Error::new(EINVAL));
}
path_owned = path.to_string();
socket.path = Some(path_owned.clone());
socket.state = State::Bound;
token = self.rng.next_u64();
socket.issued_token = Some(token);
}
socket.path = Some(path.clone());
socket.state = State::Bound;
self.socket_paths.insert(path, socket_rc.clone());
self.socket_paths.insert(path_owned, socket_rc.clone());
self.socket_tokens.insert(token, socket_rc);
Ok(0)
}
fn handle_connect(&mut self, id: usize, path_buf: &[u8]) -> Result<usize> {
let path = path_buf_to_string(path_buf)?;
fn handle_connect(&mut self, id: usize, token_buf: &[u8]) -> Result<usize> {
let token = read_num::<u64>(token_buf)?;
{
let target_rc = self
.socket_tokens
.get(&token)
.ok_or(Error::new(ECONNREFUSED))?;
let target_socket_token = target_rc
.borrow()
.issued_token
.ok_or(Error::new(ECONNREFUSED))?;
if target_socket_token != token {
return Err(Error::new(EACCES));
}
let target_rc = self
.socket_paths
.get(&path)
.ok_or(Error::new(ECONNREFUSED))?;
let target_id = target_rc.borrow().primary_id;
let target_id = target_rc.borrow().primary_id;
let socket_rc = self.get_socket(id)?;
socket_rc.borrow_mut().peer = Some(target_id);
let socket_rc = self.get_socket(id)?;
socket_rc.borrow_mut().peer = Some(target_id);
}
Ok(0)
}
@@ -257,10 +285,9 @@ impl<'sock> UdsDgramScheme<'sock> {
Ok(value_slice.len())
}
_ => {
log::warn!(
eprintln!(
"handle_setsockopt(id: {}): Unsupported option: {}",
id,
option
id, option
);
Err(Error::new(ENOPROTOOPT))
}
@@ -272,7 +299,7 @@ impl<'sock> UdsDgramScheme<'sock> {
SO_DOMAIN => {
payload.fill(0);
if payload.len() < mem::size_of::<i32>() {
log::error!(
eprintln!(
"handle_getsockopt(id: {}): SO_DOMAIN payload buffer is too small. len: {}",
id,
payload.len()
@@ -284,10 +311,9 @@ impl<'sock> UdsDgramScheme<'sock> {
Ok(domain.len())
}
_ => {
log::warn!(
eprintln!(
"handle_getsockopt(id: {}): Unsupported option: {}",
id,
option
id, option
);
Err(Error::new(ENOPROTOOPT))
}
@@ -296,7 +322,7 @@ impl<'sock> UdsDgramScheme<'sock> {
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);
eprintln!("handle_sendmsg(id: {}): msg_stream is empty.", id);
return Err(Error::new(EINVAL));
}
@@ -307,24 +333,35 @@ impl<'sock> UdsDgramScheme<'sock> {
};
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)?;
let bytes_written = Self::sendmsg_inner(
self.proc_creds_capability,
&mut remote_rc.borrow_mut(),
name,
msg_stream,
ctx,
)?;
self.post_fevent(remote_id, EVENT_READ.bits())?;
Ok(bytes_written)
}
fn sendmsg_inner(
cap_fd: usize,
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.");
eprintln!("sendmsg_inner: msg_stream is empty.");
return Err(Error::new(EINVAL));
}
let message = DataPacket::from_stream(msg_stream, name, ctx)?;
let (pid, uid, gid) = get_uid_gid_from_pid(cap_fd, ctx.pid)?;
let message = DataPacket::from_stream(
msg_stream,
name,
Credential::new(pid as i32, uid as i32, gid as i32),
)?;
let payload_len = message.len();
socket.messages.push_back(message);
@@ -360,6 +397,43 @@ impl<'sock> UdsDgramScheme<'sock> {
)
}
fn handle_unbind(&mut self, id: usize) -> Result<usize> {
let path_opt = {
let socket_rc = self.get_socket(id)?;
let mut socket = socket_rc.borrow_mut();
if socket.state != State::Bound {
return Err(Error::new(EINVAL));
}
socket.state = State::Unbound;
socket.path.take()
};
if let Some(path) = path_opt {
self.socket_paths.remove(&path);
}
Ok(0)
}
fn handle_get_token(&self, id: usize, payload: &mut [u8]) -> Result<usize> {
let socket_rc = self.get_socket(id)?;
let Some(token) = socket_rc.borrow().issued_token else {
return Err(Error::new(EINVAL));
};
let token_bytes = token.to_le_bytes();
let token_bytes_len = token_bytes.len();
if payload.len() < token_bytes_len {
eprintln!(
"handle_get_token(id: {}): Payload buffer is too small for token.",
id
);
return Err(Error::new(ENOBUFS));
}
payload[..token_bytes_len].copy_from_slice(&token_bytes);
return Ok(token_bytes_len);
}
fn handle_connect_socketpair(&mut self, id: usize) -> Result<OpenResult> {
let new_id = self.next_id;
let mut new = Socket::default();
@@ -367,7 +441,7 @@ impl<'sock> UdsDgramScheme<'sock> {
let socket_rc = self.get_socket(id)?;
if socket_rc.borrow().state == State::Closed {
log::warn!(
eprintln!(
"handle_connect_socketpair(id: {}): Attempting to connect from a closed socket.",
id
);
@@ -488,7 +562,7 @@ impl<'sock> UdsDgramScheme<'sock> {
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);
eprintln!("sendfd_inner: obtain_fd failed with error: {:?}", e);
return Err(e);
}
let socket_id = sendfd_request.id();
@@ -505,7 +579,7 @@ impl<'sock> SchemeSync for UdsDgramScheme<'sock> {
let new_id = if path.is_empty() {
self.handle_unnamed_socket(flags)
} else {
log::error!(
eprintln!(
"open(path: '{}'): Attempting to open a named socket, which is not supported.",
path
);
@@ -554,7 +628,7 @@ impl<'sock> SchemeSync for UdsDgramScheme<'sock> {
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);
eprintln!("fpath(id: {}): Socket is unnamed, has no path.", id);
Error::new(ENOENT)
})?;
Ok(Self::fpath_inner(path, buf)?)
@@ -589,6 +663,9 @@ impl<'sock> SchemeSync for UdsDgramScheme<'sock> {
if let Some(path) = path {
self.socket_paths.remove(&path);
}
if let Some(token) = socket.issued_token {
self.socket_tokens.remove(&token);
}
}
}
@@ -605,9 +682,20 @@ impl<'sock> SchemeSync for UdsDgramScheme<'sock> {
Ok(0)
}
_ => {
log::error!("fcntl(id: {}): Unsupported cmd: {}", id, cmd);
eprintln!("fcntl(id: {}): Unsupported cmd: {}", id, cmd);
Err(Error::new(EINVAL))
}
}
}
fn fstat(&mut self, id: usize, stat: &mut Stat, ctx: &CallerCtx) -> Result<()> {
self.get_socket(id)?;
*stat = Stat {
st_mode: MODE_SOCK,
..Default::default()
};
Ok(())
}
}
ipcd/src/uds/mod.rs
+40 -36
View File
@@ -1,5 +1,4 @@
use redox_rt::protocol::ProcMeta;
use redox_scheme::CallerCtx;
use std::{cmp, convert::TryInto, mem};
use syscall::{error::*, Error};
@@ -99,32 +98,27 @@ impl DataPacket {
self.payload.len()
}
fn from_stream(stream: &[u8], name: Option<String>, ctx: &CallerCtx) -> Result<Self> {
fn from_stream(stream: &[u8], name: Option<String>, cred: Credential) -> 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);
eprintln!("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),
);
let mut message = Self::new(payload.to_vec(), AncillaryData::new(cred, 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);
eprintln!("Message::from_stream: Stream too short for ancillary data. Expected len {}, actual remaining len {}", cmsg_header.data_len, stream.len() - cursor);
Error::new(EINVAL)
})?;
@@ -135,10 +129,9 @@ impl DataPacket {
message.ancillary_data.num_fds += num_fds;
}
_ => {
log::warn!(
eprintln!(
"Message::from_stream: Unsupported cmsg type received. level: {}, type: {}",
cmsg_header.level,
cmsg_header.c_type
cmsg_header.level, cmsg_header.c_type
);
return Err(Error::new(EOPNOTSUPP));
}
@@ -165,7 +158,7 @@ impl NumFromBytes for i32 {
.get(..mem::size_of::<i32>())
.and_then(|slice| slice.try_into().ok())
.ok_or_else(|| {
log::error!(
eprintln!(
"read_num: buffer is too short to read num len: {}",
buffer.len()
);
@@ -181,7 +174,23 @@ impl NumFromBytes for u32 {
.get(..mem::size_of::<u32>())
.and_then(|slice| slice.try_into().ok())
.ok_or_else(|| {
log::error!(
eprintln!(
"read_num: buffer is too short to read num len: {}",
buffer.len()
);
Error::new(EINVAL)
})?,
))
}
}
impl NumFromBytes for u64 {
fn from_le_bytes_slice(buffer: &[u8]) -> Result<Self> {
Ok(u64::from_le_bytes(
buffer
.get(..mem::size_of::<u64>())
.and_then(|slice| slice.try_into().ok())
.ok_or_else(|| {
eprintln!(
"read_num: buffer is too short to read num len: {}",
buffer.len()
);
@@ -197,7 +206,7 @@ impl NumFromBytes for usize {
.get(..mem::size_of::<usize>())
.and_then(|slice| slice.try_into().ok())
.ok_or_else(|| {
log::error!(
eprintln!(
"read_num: buffer is too short to read num len: {}",
buffer.len()
);
@@ -207,16 +216,12 @@ impl NumFromBytes for usize {
}
}
fn get_uid_gid_from_pid(target_pid: usize) -> Result<(u32, u32, u32)> {
// broken anyways
return Err(Error::new(EINVAL));
/*
fn get_uid_gid_from_pid(cap_fd: usize, 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!(
let _ = libredox::call::get_proc_credentials(cap_fd, target_pid, &mut buffer).map_err(|e| {
eprintln!(
"Failed to get process credentials for pid {}: {:?}",
target_pid,
e
target_pid, e
);
Error::new(EINVAL)
})?;
@@ -227,12 +232,11 @@ fn get_uid_gid_from_pid(target_pid: usize) -> Result<(u32, u32, u32)> {
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!(
eprintln!(
"get_msghdr_info: stream buffer is too small to read headers. len: {}",
stream.len()
);
@@ -277,7 +281,7 @@ impl<'a> MsgWriter<'a> {
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());
eprintln!("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(());
}
@@ -287,7 +291,7 @@ impl<'a> MsgWriter<'a> {
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);
eprintln!("MsgWriter::write_name: Name will be truncated. Full length: {}, buffer available: {}", name.len(), copy_len);
}
let name_len = name_write_fn(
&name,
@@ -309,7 +313,7 @@ impl<'a> MsgWriter<'a> {
.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());
eprintln!("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);
};
@@ -318,13 +322,13 @@ impl<'a> MsgWriter<'a> {
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);
eprintln!("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());
eprintln!("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);
};
@@ -337,11 +341,11 @@ impl<'a> MsgWriter<'a> {
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.");
eprintln!("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);
eprintln!("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();
@@ -356,7 +360,7 @@ impl<'a> MsgWriter<'a> {
cursor += CMSG_DATA_LEN_SIZE;
if remaining_buf.len() < cursor + data_len {
log::warn!(
eprintln!(
"CmsgWriter::write_cmsg: Not enough space for cmsg data. remaining: {}, needed: {}",
remaining_buf.len(),
cursor + data_len
@@ -390,10 +394,10 @@ impl<'a> MsgWriter<'a> {
}
}
fn path_buf_to_string(path_buf: &[u8]) -> Result<String> {
fn path_buf_to_str(path_buf: &[u8]) -> Result<&str> {
match std::str::from_utf8(path_buf) {
Ok("") => Err(Error::new(EINVAL)),
Ok(s) => Ok(s.to_string()),
Ok(s) => Ok(s),
Err(_) => Err(Error::new(EINVAL)),
}
}
ipcd/src/uds/stream.rs
+153 -74
View File
@@ -1,11 +1,12 @@
//! 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,
get_uid_gid_from_pid, path_buf_to_str, read_msghdr_info, read_num, AncillaryData, Credential,
DataPacket, MsgWriter, MIN_RECV_MSG_LEN,
};
use libc::{AF_UNIX, SO_DOMAIN, SO_PASSCRED};
use rand::prelude::*;
use redox_rt::protocol::SocketCall;
use redox_scheme::{
scheme::SchemeSync, CallerCtx, OpenResult, Response, SendFdRequest, SignalBehavior,
@@ -18,7 +19,7 @@ use std::{
mem,
rc::Rc,
};
use syscall::{error::*, flag::*, schemev2::NewFdFlags, Error};
use syscall::{error::*, flag::*, schemev2::NewFdFlags, Error, Stat};
#[derive(Debug, Default, Clone, PartialEq, Eq)]
struct Connection {
@@ -39,7 +40,7 @@ impl Connection {
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);
eprintln!("Connection::drop_fds: Attempted to drop FD #{} of {}, but fd queue is empty. State inconsistency.", i + 1, num_fd);
return Err(Error::new(EPROTO));
}
}
@@ -103,7 +104,7 @@ impl Connection {
num_fds += ancillary_data.num_fds;
}
if !msg_writer.write_rights(num_fds) {
log::warn!(
eprintln!(
"serialize_to_msgstream: Buffer too small for SCM_RIGHTS, dropping {} FDs.",
num_fds
);
@@ -123,7 +124,7 @@ impl Connection {
success
}
_ => {
log::warn!(
eprintln!(
"serialize_to_msgstream: Unsupported socket option for serialization: {}",
option
);
@@ -131,7 +132,7 @@ impl Connection {
}
};
if !result {
log::warn!("serialize_to_msgstream: Buffer too small for ancillary data, stopping further serialization.");
eprintln!("serialize_to_msgstream: Buffer too small for ancillary data, stopping further serialization.");
break;
}
}
@@ -166,6 +167,7 @@ pub struct Socket {
state: State,
awaiting: VecDeque<usize>,
connection: Option<Connection>,
issued_token: Option<u64>,
}
impl Socket {
@@ -190,7 +192,7 @@ impl Socket {
fn accept(&mut self, primary_id: usize, awaiting_client_id: usize) -> Result<Self> {
if !self.is_listening() {
log::error!(
eprintln!(
"accept(id: {}): Accept called on a non-listening socket.",
self.primary_id
);
@@ -208,10 +210,9 @@ impl Socket {
fn establish(&mut self, peer: usize) -> Result<()> {
if self.state != State::Connecting {
log::error!(
eprintln!(
"establish(id: {}): Cannot establish connection in state: {:?}",
self.primary_id,
self.state
self.primary_id, self.state
);
return Err(Error::new(EINVAL));
}
@@ -256,7 +257,7 @@ impl Socket {
if let Some(connection) = &mut self.connection {
Ok(connection)
} else {
log::error!(
eprintln!(
"Socket (id: {}): connection is None in require_connection",
self.primary_id
);
@@ -274,10 +275,9 @@ impl Socket {
fn start_listening(&mut self) -> Result<()> {
if !matches!(self.state, State::Unbound | State::Bound) {
log::error!(
eprintln!(
"start_listening(id: {}): Socket cannot listen in state {:?}.",
self.primary_id,
self.state
self.primary_id, self.state
);
return Err(Error::new(EINVAL));
}
@@ -301,17 +301,26 @@ pub struct UdsStreamScheme<'sock> {
sockets: HashMap<usize, Rc<RefCell<Socket>>>,
next_id: usize,
socket_paths: HashMap<String, Rc<RefCell<Socket>>>,
socket_tokens: HashMap<u64, Rc<RefCell<Socket>>>,
socket: &'sock SchemeSocket,
proc_creds_capability: usize,
rng: SmallRng,
}
impl<'sock> UdsStreamScheme<'sock> {
pub fn new(socket: &'sock SchemeSocket) -> Self {
Self {
pub fn new(socket: &'sock SchemeSocket) -> Result<Self> {
Ok(Self {
sockets: HashMap::new(),
next_id: 0,
socket_paths: HashMap::new(),
socket_tokens: HashMap::new(),
socket,
}
proc_creds_capability: syscall::open(
"/scheme/proc/proc-creds-capability",
syscall::O_RDONLY,
)?,
rng: SmallRng::from_entropy(),
})
}
fn post_fevent(&self, id: usize, flags: usize) -> Result<()> {
@@ -335,15 +344,14 @@ impl<'sock> UdsStreamScheme<'sock> {
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);
eprintln!("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!(
eprintln!(
"get_connected_peer(id: {}): Attempted to interact with a closed peer (id: {}).",
id,
remote_id
id, remote_id
);
return Err(Error::new(ECONNREFUSED));
}
@@ -367,7 +375,7 @@ impl<'sock> UdsStreamScheme<'sock> {
ctx: &CallerCtx,
) -> Result<usize> {
let Some(verb) = SocketCall::try_from_raw(metadata[0] as usize) else {
log::error!("call_inner: Invalid verb in metadata: {:?}", metadata);
eprintln!("call_inner: Invalid verb in metadata: {:?}", metadata);
return Err(Error::new(EINVAL));
};
match verb {
@@ -377,34 +385,43 @@ impl<'sock> UdsStreamScheme<'sock> {
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),
SocketCall::Unbind => self.handle_unbind(id),
SocketCall::GetToken => self.handle_get_token(id, payload),
_ => Err(Error::new(EOPNOTSUPP)),
}
}
fn handle_bind(&mut self, id: usize, path_buf: &[u8]) -> Result<usize> {
let path = path_buf_to_string(path_buf)?;
let path = path_buf_to_str(path_buf)?;
if self.socket_paths.contains_key(&path) {
log::error!("handle_bind: Path '{}' is already in use.", path);
if self.socket_paths.contains_key(path) {
eprintln!("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();
let path_owned: String;
let token: u64;
{
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));
if socket.state != State::Unbound {
eprintln!(
"handle_bind(id: {}): Socket is already bound or connected (state: {:?})",
id, socket.state
);
return Err(Error::new(EINVAL));
}
path_owned = path.to_string();
socket.path = Some(path_owned.clone());
socket.state = State::Bound;
token = self.rng.next_u64();
socket.issued_token.insert(token);
}
socket.path = Some(path.clone());
socket.state = State::Bound;
self.socket_paths.insert(path, socket_rc.clone());
self.socket_paths.insert(path_owned, socket_rc.clone());
self.socket_tokens.insert(token, socket_rc);
Ok(0)
}
@@ -434,15 +451,16 @@ impl<'sock> UdsStreamScheme<'sock> {
// 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<usize> {
let path = path_buf_to_string(path_buf)?;
fn handle_connect(&mut self, id: usize, token_buf: &[u8]) -> Result<usize> {
let token = read_num::<u64>(token_buf)?;
let (listener_id, flags) = {
let listener_rc = self
.socket_paths
.get(&path)
.ok_or_else(|| Error::new(ECONNREFUSED))?;
.socket_tokens
.get(&token)
.ok_or_else(|| Error::new(ECONNREFUSED))?
.clone();
let client_rc = self.get_socket(id)?;
let client_rc = self.get_socket(id)?.clone();
let mut client = client_rc.borrow_mut();
match client.state {
@@ -501,10 +519,9 @@ impl<'sock> UdsStreamScheme<'sock> {
Ok(value_slice.len())
}
_ => {
log::warn!(
eprintln!(
"socket_setsockopt(id: {}): Unsupported option: {}",
id,
option
id, option
);
Err(Error::new(ENOPROTOOPT))
}
@@ -516,7 +533,7 @@ impl<'sock> UdsStreamScheme<'sock> {
SO_DOMAIN => {
payload.fill(0);
if payload.len() < mem::size_of::<i32>() {
log::error!(
eprintln!(
"socket_getsockopt(id: {}): SO_DOMAIN payload buffer is too small. len: {}",
id,
payload.len()
@@ -528,10 +545,9 @@ impl<'sock> UdsStreamScheme<'sock> {
Ok(domain.len())
}
_ => {
log::warn!(
eprintln!(
"socket_getsockopt(id: {}): Unsupported option: {}",
id,
option
id, option
);
Err(Error::new(ENOPROTOOPT))
}
@@ -540,32 +556,44 @@ impl<'sock> UdsStreamScheme<'sock> {
fn handle_sendmsg(&mut self, id: usize, msg_stream: &[u8], ctx: &CallerCtx) -> Result<usize> {
if msg_stream.is_empty() {
log::error!("msg_stream is empty, returning EINVAL.");
eprintln!("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)?;
let bytes_written = Self::sendmsg_inner(
self.proc_creds_capability,
&mut remote_rc.borrow_mut(),
name,
msg_stream,
ctx,
)?;
self.post_fevent(remote_id, EVENT_READ.bits())?;
Ok(bytes_written)
}
fn sendmsg_inner(
cap_fd: usize,
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.");
eprintln!("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 (pid, uid, gid) = get_uid_gid_from_pid(cap_fd, ctx.pid)?;
let packet = DataPacket::from_stream(
msg_stream,
name,
Credential::new(pid as i32, uid as i32, gid as i32),
)?;
let payload_len = packet.len();
connection.packets.push_back(packet);
@@ -583,7 +611,7 @@ impl<'sock> UdsStreamScheme<'sock> {
return Self::write_eof(msg_stream);
}
State::Listening => {
log::warn!("socket_recvmsg: Called on a listening socket, returning EOPNOTSUPP.");
eprintln!("socket_recvmsg: Called on a listening socket, returning EOPNOTSUPP.");
return Err(Error::new(EOPNOTSUPP));
}
_ => return Err(Error::new(ENOTCONN)),
@@ -605,7 +633,7 @@ impl<'sock> UdsStreamScheme<'sock> {
fn write_eof(buffer: &mut [u8]) -> Result<usize> {
// 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.");
eprintln!("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
@@ -624,6 +652,44 @@ impl<'sock> UdsStreamScheme<'sock> {
Ok(written_len)
}
fn handle_unbind(&mut self, id: usize) -> Result<usize> {
let path_opt = {
let socket_rc = self.get_socket(id)?;
let mut socket = socket_rc.borrow_mut();
if socket.state != State::Bound {
return Err(Error::new(EINVAL));
}
socket.state = State::Unbound;
socket.path.take()
};
if let Some(path) = path_opt {
self.socket_paths.remove(&path);
}
Ok(0)
}
fn handle_get_token(&self, id: usize, payload: &mut [u8]) -> Result<usize> {
let socket_rc = self.get_socket(id)?;
let Some(token) = socket_rc.borrow().issued_token else {
return Err(Error::new(EINVAL));
};
let token_bytes = token.to_le_bytes();
let token_bytes_len = token_bytes.len();
if payload.len() < token_bytes_len {
eprintln!(
"handle_get_token(id: {}): Payload buffer is too small for token.",
id
);
return Err(Error::new(ENOBUFS));
}
payload[..token_bytes_len].copy_from_slice(&token_bytes);
return Ok(token_bytes_len);
}
fn accept_connection(
&mut self,
listener_socket: &mut Socket,
@@ -653,7 +719,7 @@ impl<'sock> UdsStreamScheme<'sock> {
fn handle_accept(&mut self, id: usize, socket: &mut Socket) -> Result<Option<OpenResult>> {
let flags = socket.flags;
if !socket.is_listening() {
log::error!(
eprintln!(
"socket_accept: Socket state is not Listening for id: {}",
id
);
@@ -681,7 +747,7 @@ impl<'sock> UdsStreamScheme<'sock> {
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);
eprintln!("handle_start_listening: Path '{}' is already in use.", path);
return Err(Error::new(EADDRINUSE));
}
}
@@ -720,7 +786,7 @@ impl<'sock> UdsStreamScheme<'sock> {
let mut socket = socket_rc.borrow_mut();
if socket.state == State::Closed {
log::error!(
eprintln!(
"socket_connect_socketpair: Base socket {} is already closed.",
id
);
@@ -790,7 +856,7 @@ impl<'sock> UdsStreamScheme<'sock> {
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);
eprintln!("sendfd_inner: obtain_fd failed with error: {:?}", e);
return Err(e);
}
let receiver_rc = self.get_socket(receiver_id)?;
@@ -856,6 +922,10 @@ impl<'sock> UdsStreamScheme<'sock> {
self.socket_paths.remove(path);
}
if let Some(token) = &socket.issued_token {
self.socket_tokens.remove(&token);
}
// Notify all waiting clients about listener closure
for client_id in &socket.awaiting {
if let Ok(client_rc) = self.get_socket(*client_id) {
@@ -870,11 +940,8 @@ impl<'sock> UdsStreamScheme<'sock> {
fn handle_other_closure(&mut self, socket_rc: Rc<RefCell<Socket>>) {
// 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 mut socket = socket_rc.borrow_mut();
if matches!(socket.state, State::Established | State::Accepted) {
let Ok(connection) = socket.require_connection() else {
return;
};
@@ -889,13 +956,14 @@ impl<'sock> UdsStreamScheme<'sock> {
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);
}
if let Some(path) = socket.path.take() {
// 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;
if let Some(token) = socket.issued_token {
self.socket_tokens.remove(&token);
}
socket.state = State::Closed;
}
fn fpath_inner(path: &String, buf: &mut [u8]) -> Result<usize> {
@@ -920,7 +988,7 @@ impl<'sock> SchemeSync for UdsStreamScheme<'sock> {
let new_id = if path.is_empty() {
self.handle_unnamed_socket(flags)
} else {
log::error!(
eprintln!(
"open(path: '{}'): Attempting to open a named socket, which is not supported.",
path
);
@@ -1032,4 +1100,15 @@ impl<'sock> SchemeSync for UdsStreamScheme<'sock> {
_ => Err(Error::new(EINVAL)),
}
}
fn fstat(&mut self, id: usize, stat: &mut Stat, ctx: &CallerCtx) -> Result<()> {
self.get_socket(id)?;
*stat = Stat {
st_mode: MODE_SOCK,
..Default::default()
};
Ok(())
}
}