use std::ffi::{CStr, OsStr, c_int, c_void};
use std::fs::File;
use std::io::{Read, Write};
use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, OwnedFd};
use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::FileTypeExt;
use std::os::unix::net::UnixStream;
use std::time::Instant;
use super::auth_adapter::auth_worker_spawn::{
InstalledAuthWorkerImage, spawn_installed_auth_worker,
};
use super::auth_adapter::broker_plan::{
AcknowledgedBrokerLaunchPlan, BROKER_ACK_BYTES, BROKER_PLAN_PREFIX_BYTES,
ExactParentBrokerLaunchPlan, MAX_BROKER_PLAN_BYTES, ReceivedBrokerLaunchPlan, broker_plan_ack,
parse_broker_plan_prefix,
};
use super::auth_adapter::broker_report::finish_broker_trace_report;
#[cfg(test)]
use super::auth_adapter::broker_report::{BROKER_RESUME_BYTE, encode_broker_trace_report};
use super::auth_adapter::broker_spawn::{
BROKER_CONTROL_FD, BROKER_GATE_FD, BROKER_TRACE_FD, INSTALLED_BROKER_MODE,
INSTALLED_CONTROL_ARGUMENT, INSTALLED_GATE_ARGUMENT, INSTALLED_TRACE_ARGUMENT, START_BYTE,
};
use super::auth_adapter::{
AuthWorkerPool, DedicatedChildWaitDomain, DirectChildAuthWorkerAuthority, FreshAuthJobId,
FreshAuthWorkerGeneration,
};
use super::{SupervisorWireError, is_deployer_helper_path};
#[path = "supervisor_broker_launcher.rs"]
pub(super) mod broker_launcher;
use crate::backend::macos::bootstrap::random_nonce;
use broker_launcher::{InstalledLauncherImage, spawn_fixed_launcher};
const F_GETFD: c_int = 1;
const F_SETFD: c_int = 2;
const F_GETFL: c_int = 3;
const F_SETFL: c_int = 4;
const FD_CLOEXEC: c_int = 1;
const O_ACCMODE: c_int = 3;
const O_RDONLY: c_int = 0;
const O_RDWR: c_int = 2;
const O_NONBLOCK: c_int = 0x0000_0004;
const EAGAIN: c_int = 35;
const EINTR: c_int = 4;
const POLLIN: i16 = 0x0001;
const POLLOUT: i16 = 0x0004;
const POLLERR: i16 = 0x0008;
const POLLHUP: i16 = 0x0010;
const POLLNVAL: i16 = 0x0020;
const SOL_SOCKET: c_int = 0xffff;
const SO_TYPE: c_int = 0x1008;
const SOCK_STREAM: c_int = 1;
#[repr(C)]
struct PollFd {
fd: c_int,
events: i16,
revents: i16,
}
unsafe extern "C" {
fn _exit(status: c_int) -> !;
fn fcntl(fd: c_int, command: c_int, ...) -> c_int;
fn getsockopt(
fd: c_int,
level: c_int,
option: c_int,
value: *mut c_void,
length: *mut u32,
) -> c_int;
fn poll(descriptors: *mut PollFd, count: u32, timeout_ms: c_int) -> c_int;
fn read(fd: c_int, buffer: *mut u8, count: usize) -> isize;
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) enum BrokerEntryError {
InvalidArguments,
InvalidGate,
InvalidControl,
InvalidTrace,
Descriptor(c_int),
Read(c_int),
InvalidActivation,
Plan(SupervisorWireError),
Control(c_int),
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) enum BrokerGateExit {
ServiceGoneBeforeActivation,
ServiceGone,
}
#[must_use = "a dormant broker gate must be consumed or closed"]
#[derive(Debug)]
pub(super) struct DormantBrokerGate {
reader: OwnedFd,
}
pub(super) struct DormantBrokerProcess {
gate: DormantBrokerGate,
control: UnixStream,
trace: UnixStream,
}
pub(super) struct StagedDormantBroker {
gate: DormantBrokerGate,
plan: AcknowledgedBrokerLaunchPlan,
trace: UnixStream,
}
pub(super) struct ActiveBrokerProcess {
pub(super) gate: ActiveBrokerGate,
pub(super) plan: ExactParentBrokerLaunchPlan,
trace: UnixStream,
}
#[cfg(test)]
pub(super) struct ReportedActiveBroker {
pub(super) gate: ActiveBrokerGate,
pub(super) plan: ExactParentBrokerLaunchPlan,
trace: UnixStream,
}
#[cfg(test)]
pub(super) struct ResumedActiveBroker {
pub(super) gate: ActiveBrokerGate,
pub(super) plan: ExactParentBrokerLaunchPlan,
}
#[must_use = "an active broker must retain and monitor service death"]
#[derive(Debug)]
pub(super) struct ActiveBrokerGate {
reader: OwnedFd,
}
impl DormantBrokerGate {
pub(super) unsafe fn adopt_fixed_process(
installed_path: &CStr,
) -> Result<DormantBrokerProcess, BrokerEntryError> {
validate_fixed_arguments(installed_path, std::env::args_os())?;
let gate = unsafe { Self::adopt_fixed_gate() }?;
let control =
unsafe { adopt_fixed_socket(BROKER_CONTROL_FD, BrokerEntryError::InvalidControl) }?;
let trace = unsafe { adopt_fixed_socket(BROKER_TRACE_FD, BrokerEntryError::InvalidTrace) }?;
Ok(DormantBrokerProcess {
gate,
control,
trace,
})
}
pub(super) fn wait_for_activation(
self,
) -> Result<Result<ActiveBrokerGate, BrokerGateExit>, BrokerEntryError> {
let mut activation = 0_u8;
match read_retry(self.reader.as_raw_fd(), &mut activation)? {
0 => return Ok(Err(BrokerGateExit::ServiceGoneBeforeActivation)),
1 if activation == START_BYTE[0] => {}
1 => return Err(BrokerEntryError::InvalidActivation),
_ => unreachable!("one-byte read returned an impossible length"),
}
set_nonblocking(self.reader.as_raw_fd(), true)?;
let mut extra = 0_u8;
let probe = read_once(self.reader.as_raw_fd(), &mut extra);
match probe {
Ok(0) => Ok(Err(BrokerGateExit::ServiceGone)),
Ok(1) => Err(BrokerEntryError::InvalidActivation),
Ok(_) => unreachable!("one-byte read returned an impossible length"),
Err(error) if error == EAGAIN => {
set_nonblocking(self.reader.as_raw_fd(), false)?;
Ok(Ok(ActiveBrokerGate {
reader: self.reader,
}))
}
Err(error) if error == EINTR => {
set_nonblocking(self.reader.as_raw_fd(), false)?;
let active = ActiveBrokerGate {
reader: self.reader,
};
active.reject_extra_or_confirm_live()
}
Err(error) => Err(BrokerEntryError::Read(error)),
}
}
unsafe fn adopt_fixed_gate() -> Result<Self, BrokerEntryError> {
if unsafe { fcntl(BROKER_GATE_FD, F_GETFD) } < 0 {
return Err(BrokerEntryError::InvalidGate);
}
let reader = unsafe { OwnedFd::from_raw_fd(BROKER_GATE_FD) };
let flags = unsafe { fcntl(reader.as_raw_fd(), F_GETFL) };
if flags < 0 {
return Err(BrokerEntryError::Descriptor(last_errno()));
}
if flags & O_ACCMODE != O_RDONLY {
return Err(BrokerEntryError::InvalidGate);
}
let file = File::from(reader);
let metadata = file
.metadata()
.map_err(|error| BrokerEntryError::Descriptor(error.raw_os_error().unwrap_or(0)))?;
if !metadata.file_type().is_fifo() {
return Err(BrokerEntryError::InvalidGate);
}
let reader = unsafe { OwnedFd::from_raw_fd(file.into_raw_fd()) };
if unsafe { fcntl(reader.as_raw_fd(), F_SETFD, FD_CLOEXEC) } != 0 {
return Err(BrokerEntryError::Descriptor(last_errno()));
}
set_nonblocking(reader.as_raw_fd(), false)?;
Ok(Self { reader })
}
}
impl DormantBrokerProcess {
#[cfg(test)]
pub(in crate::backend::macos::supervisor) unsafe fn adopt_test_channels()
-> Result<Self, BrokerEntryError> {
if unsafe { fcntl(BROKER_GATE_FD, F_GETFD) } < 0 {
return Err(BrokerEntryError::InvalidGate);
}
let reader = unsafe { OwnedFd::from_raw_fd(BROKER_GATE_FD) };
set_nonblocking(reader.as_raw_fd(), false)?;
let gate = DormantBrokerGate { reader };
if unsafe { fcntl(BROKER_CONTROL_FD, F_GETFD) } < 0 {
return Err(BrokerEntryError::InvalidControl);
}
let control = UnixStream::from(unsafe { OwnedFd::from_raw_fd(BROKER_CONTROL_FD) });
control
.set_nonblocking(true)
.map_err(|error| BrokerEntryError::Descriptor(error.raw_os_error().unwrap_or(0)))?;
if unsafe { fcntl(BROKER_TRACE_FD, F_GETFD) } < 0 {
return Err(BrokerEntryError::InvalidTrace);
}
let trace = UnixStream::from(unsafe { OwnedFd::from_raw_fd(BROKER_TRACE_FD) });
trace
.set_nonblocking(true)
.map_err(|error| BrokerEntryError::Descriptor(error.raw_os_error().unwrap_or(0)))?;
Ok(Self {
gate,
control,
trace,
})
}
pub(super) fn stage_plan(
mut self,
) -> Result<Result<StagedDormantBroker, BrokerGateExit>, BrokerEntryError> {
set_nonblocking(self.gate.reader.as_raw_fd(), true)?;
let mut outer = [0_u8; 4];
if let Some(exit) = read_control_while_dormant(
&mut self.control,
self.gate.reader.as_raw_fd(),
&mut outer,
None,
)? {
return Ok(Err(exit));
}
let frame_len = usize::try_from(u32::from_le_bytes(outer))
.map_err(|_| BrokerEntryError::Plan(SupervisorWireError::LimitExceeded))?;
if !(256..=MAX_BROKER_PLAN_BYTES).contains(&frame_len) {
return Err(BrokerEntryError::Plan(SupervisorWireError::LimitExceeded));
}
let mut frame = vec![0_u8; frame_len];
let (prefix, remaining) = frame.split_at_mut(BROKER_PLAN_PREFIX_BYTES);
if let Some(exit) = read_control_while_dormant(
&mut self.control,
self.gate.reader.as_raw_fd(),
prefix,
None,
)? {
return Ok(Err(exit));
}
let prefix: &[u8; BROKER_PLAN_PREFIX_BYTES] = (&*prefix)
.try_into()
.map_err(|_| BrokerEntryError::Plan(SupervisorWireError::Malformed))?;
let parsed = parse_broker_plan_prefix(prefix, frame_len).map_err(BrokerEntryError::Plan)?;
if parsed.frame_len != frame_len {
return Err(BrokerEntryError::Plan(SupervisorWireError::Malformed));
}
if let Some(exit) = read_control_while_dormant(
&mut self.control,
self.gate.reader.as_raw_fd(),
remaining,
Some(parsed.deadline.local()),
)? {
return Ok(Err(exit));
}
if let Some(exit) = require_control_frame_eof(
&mut self.control,
self.gate.reader.as_raw_fd(),
parsed.deadline.local(),
)? {
return Ok(Err(exit));
}
let received = ReceivedBrokerLaunchPlan::decode_with_deadline(&frame, parsed.deadline)
.map_err(BrokerEntryError::Plan)?;
ensure_deadline_live(Some(received.deadline().local()))?;
let ack = broker_plan_ack(&frame);
debug_assert_eq!(ack.len(), BROKER_ACK_BYTES);
if let Some(exit) = write_control_while_dormant(
&mut self.control,
self.gate.reader.as_raw_fd(),
&ack,
received.deadline().local(),
)? {
return Ok(Err(exit));
}
ensure_deadline_live(Some(received.deadline().local()))?;
let plan = unsafe { received.acknowledge_exact_parent() };
drop(self.control);
Ok(Ok(StagedDormantBroker {
gate: self.gate,
plan,
trace: self.trace,
}))
}
}
impl StagedDormantBroker {
pub(super) fn wait_for_activation(
self,
) -> Result<Result<ActiveBrokerProcess, BrokerGateExit>, BrokerEntryError> {
let deadline = self.plan.deadline().local();
match self.gate.wait_for_activation_until(deadline)? {
Err(exit) => Ok(Err(exit)),
Ok(gate) => {
ensure_deadline_live(Some(deadline))?;
let plan = unsafe { self.plan.activate() };
Ok(Ok(ActiveBrokerProcess {
gate,
plan,
trace: self.trace,
}))
}
}
}
}
impl DormantBrokerGate {
fn wait_for_activation_until(
self,
deadline: Instant,
) -> Result<Result<ActiveBrokerGate, BrokerGateExit>, BrokerEntryError> {
set_nonblocking(self.reader.as_raw_fd(), true)?;
loop {
let mut activation = 0_u8;
match read_once(self.reader.as_raw_fd(), &mut activation) {
Ok(0) => return Ok(Err(BrokerGateExit::ServiceGoneBeforeActivation)),
Ok(1) if activation == START_BYTE[0] => break,
Ok(1) => return Err(BrokerEntryError::InvalidActivation),
Ok(_) => unreachable!("one-byte read returned impossible length"),
Err(error) if error == EINTR => continue,
Err(error) if error == EAGAIN => {
poll_gate_until(self.reader.as_raw_fd(), deadline)?
}
Err(error) => return Err(BrokerEntryError::Read(error)),
}
}
let mut extra = 0_u8;
loop {
match read_once(self.reader.as_raw_fd(), &mut extra) {
Ok(0) => return Ok(Err(BrokerGateExit::ServiceGone)),
Ok(1) => return Err(BrokerEntryError::InvalidActivation),
Ok(_) => unreachable!("one-byte read returned impossible length"),
Err(error) if error == EINTR => continue,
Err(error) if error == EAGAIN => {
set_nonblocking(self.reader.as_raw_fd(), false)?;
ensure_deadline_live(Some(deadline))?;
return Ok(Ok(ActiveBrokerGate {
reader: self.reader,
}));
}
Err(error) => return Err(BrokerEntryError::Read(error)),
}
}
}
}
fn poll_gate_until(fd: c_int, deadline: Instant) -> Result<(), BrokerEntryError> {
loop {
let remaining = deadline
.checked_duration_since(Instant::now())
.ok_or(BrokerEntryError::Plan(SupervisorWireError::LimitExceeded))?;
let mut descriptor = PollFd {
fd,
events: POLLIN,
revents: 0,
};
let timeout = c_int::try_from(remaining.as_millis()).unwrap_or(c_int::MAX);
let result = unsafe { poll(&raw mut descriptor, 1, timeout) };
if result > 0 {
return Ok(());
}
if result == 0 {
ensure_deadline_live(Some(deadline))?;
continue;
}
let error = last_errno();
if error != EINTR {
return Err(BrokerEntryError::Read(error));
}
}
}
unsafe fn adopt_fixed_socket(
fd: c_int,
invalid: BrokerEntryError,
) -> Result<UnixStream, BrokerEntryError> {
if unsafe { fcntl(fd, F_GETFD) } < 0 {
return Err(invalid);
}
let flags = unsafe { fcntl(fd, F_GETFL) };
let mut socket_type: c_int = 0;
let mut socket_type_len = u32::try_from(std::mem::size_of::<c_int>())
.map_err(|_| BrokerEntryError::InvalidControl)?;
if flags < 0
|| flags & O_ACCMODE != O_RDWR
|| unsafe {
getsockopt(
fd,
SOL_SOCKET,
SO_TYPE,
(&raw mut socket_type).cast(),
&raw mut socket_type_len,
)
} != 0
|| socket_type_len as usize != std::mem::size_of::<c_int>()
|| socket_type != SOCK_STREAM
{
return Err(invalid);
}
let owned = unsafe { OwnedFd::from_raw_fd(fd) };
let file = File::from(owned);
let metadata = file
.metadata()
.map_err(|error| BrokerEntryError::Descriptor(error.raw_os_error().unwrap_or(0)))?;
if !metadata.file_type().is_socket() {
return Err(BrokerEntryError::InvalidControl);
}
let owned = unsafe { OwnedFd::from_raw_fd(file.into_raw_fd()) };
if unsafe { fcntl(owned.as_raw_fd(), F_SETFD, FD_CLOEXEC) } != 0 {
return Err(BrokerEntryError::Descriptor(last_errno()));
}
let stream = UnixStream::from(owned);
stream
.set_nonblocking(true)
.map_err(|error| BrokerEntryError::Descriptor(error.raw_os_error().unwrap_or(0)))?;
Ok(stream)
}
fn read_control_while_dormant(
control: &mut UnixStream,
gate_fd: c_int,
mut bytes: &mut [u8],
deadline: Option<Instant>,
) -> Result<Option<BrokerGateExit>, BrokerEntryError> {
while !bytes.is_empty() {
if let Some(exit) = probe_dormant_gate(gate_fd)? {
return Ok(Some(exit));
}
ensure_deadline_live(deadline)?;
match control.read(bytes) {
Ok(0) => return Err(BrokerEntryError::Control(0)),
Ok(count) => bytes = &mut bytes[count..],
Err(ref error) if error.kind() == std::io::ErrorKind::Interrupted => {}
Err(ref error) if error.kind() == std::io::ErrorKind::WouldBlock => {
if let Some(exit) =
poll_control_and_gate(gate_fd, control.as_raw_fd(), POLLIN, deadline)?
{
return Ok(Some(exit));
}
}
Err(error) => {
return Err(BrokerEntryError::Control(error.raw_os_error().unwrap_or(0)));
}
}
}
Ok(None)
}
fn write_control_while_dormant(
control: &mut UnixStream,
gate_fd: c_int,
mut bytes: &[u8],
deadline: Instant,
) -> Result<Option<BrokerGateExit>, BrokerEntryError> {
while !bytes.is_empty() {
if let Some(exit) = probe_dormant_gate(gate_fd)? {
return Ok(Some(exit));
}
ensure_deadline_live(Some(deadline))?;
match control.write(bytes) {
Ok(0) => return Err(BrokerEntryError::Control(0)),
Ok(count) => bytes = &bytes[count..],
Err(ref error) if error.kind() == std::io::ErrorKind::Interrupted => {}
Err(ref error) if error.kind() == std::io::ErrorKind::WouldBlock => {
if let Some(exit) =
poll_control_and_gate(gate_fd, control.as_raw_fd(), POLLOUT, Some(deadline))?
{
return Ok(Some(exit));
}
}
Err(error) => {
return Err(BrokerEntryError::Control(error.raw_os_error().unwrap_or(0)));
}
}
}
Ok(None)
}
fn require_control_frame_eof(
control: &mut UnixStream,
gate_fd: c_int,
deadline: Instant,
) -> Result<Option<BrokerGateExit>, BrokerEntryError> {
let mut extra = [0_u8; 1];
loop {
if let Some(exit) = probe_dormant_gate(gate_fd)? {
return Ok(Some(exit));
}
ensure_deadline_live(Some(deadline))?;
match control.read(&mut extra) {
Ok(0) => return Ok(None),
Ok(1) => return Err(BrokerEntryError::Plan(SupervisorWireError::Malformed)),
Ok(_) => unreachable!("one-byte read returned impossible length"),
Err(ref error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
Err(ref error) if error.kind() == std::io::ErrorKind::WouldBlock => {
if let Some(exit) =
poll_control_and_gate(gate_fd, control.as_raw_fd(), POLLIN, Some(deadline))?
{
return Ok(Some(exit));
}
}
Err(error) => {
return Err(BrokerEntryError::Control(error.raw_os_error().unwrap_or(0)));
}
}
}
}
fn probe_dormant_gate(gate_fd: c_int) -> Result<Option<BrokerGateExit>, BrokerEntryError> {
let mut byte = 0_u8;
loop {
return match read_once(gate_fd, &mut byte) {
Ok(0) => Ok(Some(BrokerGateExit::ServiceGoneBeforeActivation)),
Ok(1) => Err(BrokerEntryError::InvalidActivation),
Ok(_) => unreachable!("one-byte read returned impossible length"),
Err(error) if error == EINTR => continue,
Err(error) if error == EAGAIN => Ok(None),
Err(error) => Err(BrokerEntryError::Read(error)),
};
}
}
fn ensure_deadline_live(deadline: Option<Instant>) -> Result<(), BrokerEntryError> {
if deadline.is_some_and(|value| Instant::now() >= value) {
Err(BrokerEntryError::Plan(SupervisorWireError::LimitExceeded))
} else {
Ok(())
}
}
fn poll_control_and_gate(
gate_fd: c_int,
control_fd: c_int,
control_events: i16,
deadline: Option<Instant>,
) -> Result<Option<BrokerGateExit>, BrokerEntryError> {
loop {
let timeout = match deadline {
Some(deadline) => {
let remaining = deadline
.checked_duration_since(Instant::now())
.ok_or(BrokerEntryError::Plan(SupervisorWireError::LimitExceeded))?;
c_int::try_from(remaining.as_millis()).unwrap_or(c_int::MAX)
}
None => -1,
};
let mut descriptors = [
PollFd {
fd: gate_fd,
events: POLLIN,
revents: 0,
},
PollFd {
fd: control_fd,
events: control_events,
revents: 0,
},
];
let result = unsafe { poll(descriptors.as_mut_ptr(), 2, timeout) };
if result == 0 {
if deadline.is_some_and(|value| Instant::now() >= value) {
return Err(BrokerEntryError::Plan(SupervisorWireError::LimitExceeded));
}
continue;
}
if result < 0 {
let error = last_errno();
if error == EINTR {
continue;
}
return Err(BrokerEntryError::Control(error));
}
if descriptors[0].revents != 0 {
let mut byte = 0_u8;
return match read_once(gate_fd, &mut byte) {
Ok(0) => Ok(Some(BrokerGateExit::ServiceGoneBeforeActivation)),
Ok(1) => Err(BrokerEntryError::InvalidActivation),
Ok(_) => unreachable!("one-byte read returned impossible length"),
Err(error) if error == EAGAIN || error == EINTR => continue,
Err(error) => Err(BrokerEntryError::Read(error)),
};
}
if descriptors[1].revents & control_events != 0 {
return Ok(None);
}
if descriptors[1].revents & (POLLERR | POLLHUP | POLLNVAL) != 0 {
return Err(BrokerEntryError::Control(0));
}
}
}
impl ActiveBrokerProcess {
#[cfg(test)]
pub(super) unsafe fn report_exact_trace_stops(
mut self,
) -> Result<Result<ReportedActiveBroker, BrokerGateExit>, BrokerEntryError> {
let deadline = self.plan.deadline().local();
ensure_deadline_live(Some(deadline))?;
let bytes = encode_broker_trace_report(self.plan.trace_report_binding())
.map_err(|error| BrokerEntryError::Plan(error.into()))?;
set_nonblocking(self.gate.reader.as_raw_fd(), true)?;
if let Some(_exit) = write_control_while_dormant(
&mut self.trace,
self.gate.reader.as_raw_fd(),
&bytes,
deadline,
)? {
return Ok(Err(BrokerGateExit::ServiceGone));
}
if let Some(exit) =
finish_trace_report_before_authority(&self.trace, self.gate.reader.as_raw_fd())?
{
return Ok(Err(exit));
}
Ok(Ok(ReportedActiveBroker {
gate: self.gate,
plan: self.plan,
trace: self.trace,
}))
}
#[cfg(test)]
pub(in crate::backend::macos::supervisor) fn abandon_trace_for_test(self) -> ActiveBrokerGate {
self.gate
}
}
#[cfg(test)]
impl ReportedActiveBroker {
pub(super) fn wait_for_ready_commit(
mut self,
) -> Result<Result<ResumedActiveBroker, BrokerGateExit>, BrokerEntryError> {
let mut resume = [0_u8; 1];
if read_resume_commit(&mut self.trace, self.gate.reader.as_raw_fd(), &mut resume)?.is_some()
{
return Ok(Err(BrokerGateExit::ServiceGone));
}
if resume != BROKER_RESUME_BYTE {
return Err(BrokerEntryError::Plan(SupervisorWireError::Malformed));
}
if require_resume_commit_eof(&mut self.trace, self.gate.reader.as_raw_fd())?.is_some() {
return Ok(Err(BrokerGateExit::ServiceGone));
}
if final_resume_gate_probe(self.gate.reader.as_raw_fd())?.is_some() {
return Ok(Err(BrokerGateExit::ServiceGone));
}
drop(self.trace);
set_nonblocking(self.gate.reader.as_raw_fd(), false)?;
Ok(Ok(ResumedActiveBroker {
gate: self.gate,
plan: self.plan,
}))
}
}
fn read_resume_commit(
trace: &mut UnixStream,
gate_fd: c_int,
resume: &mut [u8; 1],
) -> Result<Option<BrokerGateExit>, BrokerEntryError> {
loop {
if let Some(exit) = probe_dormant_gate(gate_fd)? {
return Ok(Some(exit));
}
match trace.read(resume) {
Ok(0) => return Err(BrokerEntryError::Plan(SupervisorWireError::Malformed)),
Ok(1) => return Ok(None),
Ok(_) => unreachable!("one-byte read returned impossible length"),
Err(ref error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
Err(ref error) if error.kind() == std::io::ErrorKind::WouldBlock => {
if let Some(exit) = poll_control_and_gate(gate_fd, trace.as_raw_fd(), POLLIN, None)?
{
return Ok(Some(exit));
}
}
Err(error) => {
return Err(BrokerEntryError::Control(error.raw_os_error().unwrap_or(0)));
}
}
}
}
fn require_resume_commit_eof(
trace: &mut UnixStream,
gate_fd: c_int,
) -> Result<Option<BrokerGateExit>, BrokerEntryError> {
let mut extra = [0_u8; 1];
loop {
if let Some(exit) = probe_dormant_gate(gate_fd)? {
return Ok(Some(exit));
}
match trace.read(&mut extra) {
Ok(0) => return Ok(None),
Ok(1) => return Err(BrokerEntryError::Plan(SupervisorWireError::Malformed)),
Ok(_) => unreachable!("one-byte read returned impossible length"),
Err(ref error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
Err(ref error) if error.kind() == std::io::ErrorKind::WouldBlock => {
if let Some(exit) = poll_control_and_gate(gate_fd, trace.as_raw_fd(), POLLIN, None)?
{
return Ok(Some(exit));
}
}
Err(error) => {
return Err(BrokerEntryError::Control(error.raw_os_error().unwrap_or(0)));
}
}
}
}
#[cfg(test)]
fn final_resume_gate_probe(gate_fd: c_int) -> Result<Option<BrokerGateExit>, BrokerEntryError> {
Ok(probe_dormant_gate(gate_fd)?.map(|_| BrokerGateExit::ServiceGone))
}
fn finish_trace_report_before_authority(
trace: &UnixStream,
gate_fd: c_int,
) -> Result<Option<BrokerGateExit>, BrokerEntryError> {
finish_broker_trace_report(trace).map_err(|error| BrokerEntryError::Plan(error.into()))?;
if probe_dormant_gate(gate_fd)?.is_some() {
return Ok(Some(BrokerGateExit::ServiceGone));
}
Ok(None)
}
impl ActiveBrokerGate {
pub(super) fn wait_for_service_death(self) -> Result<BrokerGateExit, BrokerEntryError> {
let mut unexpected = 0_u8;
match read_retry(self.reader.as_raw_fd(), &mut unexpected)? {
0 => Ok(BrokerGateExit::ServiceGone),
1 => Err(BrokerEntryError::InvalidActivation),
_ => unreachable!("one-byte read returned an impossible length"),
}
}
fn reject_extra_or_confirm_live(
self,
) -> Result<Result<Self, BrokerGateExit>, BrokerEntryError> {
set_nonblocking(self.reader.as_raw_fd(), true)?;
let mut extra = 0_u8;
loop {
match read_once(self.reader.as_raw_fd(), &mut extra) {
Ok(0) => return Ok(Err(BrokerGateExit::ServiceGone)),
Ok(1) => return Err(BrokerEntryError::InvalidActivation),
Ok(_) => unreachable!("one-byte read returned an impossible length"),
Err(error) if error == EINTR => continue,
Err(error) if error == EAGAIN => {
set_nonblocking(self.reader.as_raw_fd(), false)?;
return Ok(Ok(self));
}
Err(error) => return Err(BrokerEntryError::Read(error)),
}
}
}
#[cfg(test)]
fn descriptor(&self) -> c_int {
self.reader.as_raw_fd()
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum FixedBrokerProcessFailure {
InvalidEntry,
Setup,
Protocol,
ServiceGone,
Launcher,
}
pub(in crate::backend::macos) unsafe fn run_fixed_broker_process(
installed_path: &CStr,
launcher_path: &CStr,
auth_worker_path: &CStr,
) -> ! {
let status = match fixed_broker_process(installed_path, launcher_path, auth_worker_path) {
Ok(()) | Err(FixedBrokerProcessFailure::ServiceGone) => 0,
Err(FixedBrokerProcessFailure::InvalidEntry) => 64,
Err(FixedBrokerProcessFailure::Setup | FixedBrokerProcessFailure::Protocol) => 65,
Err(FixedBrokerProcessFailure::Launcher) => 67,
};
unsafe { _exit(status) }
}
fn fixed_broker_process(
installed_path: &CStr,
launcher_path: &CStr,
auth_worker_path: &CStr,
) -> Result<(), FixedBrokerProcessFailure> {
validate_fixed_arguments(installed_path, std::env::args_os())
.map_err(|_| FixedBrokerProcessFailure::InvalidEntry)?;
let mut wait_domain = unsafe { DedicatedChildWaitDomain::establish_at_service_startup() }
.map_err(|_| FixedBrokerProcessFailure::Setup)?;
let launcher_image =
unsafe { InstalledLauncherImage::from_verified_installation(launcher_path) }
.map_err(|_| FixedBrokerProcessFailure::Setup)?;
let auth_worker_image =
unsafe { InstalledAuthWorkerImage::from_verified_installation(auth_worker_path) }
.map_err(|_| FixedBrokerProcessFailure::Setup)?;
let dormant = unsafe { DormantBrokerGate::adopt_fixed_process(installed_path) }
.map_err(|_| FixedBrokerProcessFailure::InvalidEntry)?;
let staged = match dormant
.stage_plan()
.map_err(|_| FixedBrokerProcessFailure::Protocol)?
{
Ok(staged) => staged,
Err(BrokerGateExit::ServiceGoneBeforeActivation | BrokerGateExit::ServiceGone) => {
return Err(FixedBrokerProcessFailure::ServiceGone);
}
};
let active = match staged
.wait_for_activation()
.map_err(|_| FixedBrokerProcessFailure::Protocol)?
{
Ok(active) => active,
Err(BrokerGateExit::ServiceGoneBeforeActivation | BrokerGateExit::ServiceGone) => {
return Err(FixedBrokerProcessFailure::ServiceGone);
}
};
let random_job = random_nonce().map_err(|_| FixedBrokerProcessFailure::Setup)?;
let job_id = unsafe { FreshAuthJobId::from_fresh_random(random_job) }
.map_err(|_| FixedBrokerProcessFailure::Setup)?;
let generation = unsafe { FreshAuthWorkerGeneration::from_unique_service_value(1) }
.map_err(|_| FixedBrokerProcessFailure::Setup)?;
let worker = spawn_installed_auth_worker(&auth_worker_image, generation, &mut wait_domain)
.map_err(|_| FixedBrokerProcessFailure::Setup)?;
let mut pool: AuthWorkerPool<DirectChildAuthWorkerAuthority> =
AuthWorkerPool::from_spawned_workers(vec![worker])
.map_err(|_| FixedBrokerProcessFailure::Setup)?;
let spawned = spawn_fixed_launcher(active, &launcher_image, &mut wait_domain)
.map_err(|_| FixedBrokerProcessFailure::Launcher)?;
let initial = spawned
.wait_initial_stop()
.map_err(|_| FixedBrokerProcessFailure::Launcher)?;
let mut awaiting = initial
.prove_trace_and_continue_to_exec()
.map_err(|_| FixedBrokerProcessFailure::Launcher)?;
awaiting
.deliver_plan()
.map_err(|_| FixedBrokerProcessFailure::Launcher)?;
let held = awaiting
.wait_exec_trap()
.map_err(|_| FixedBrokerProcessFailure::Launcher)?;
let verified = held
.verify_signature(&mut pool, job_id)
.map_err(|_| FixedBrokerProcessFailure::Launcher)?;
let reported = match verified
.report_trace_stops()
.map_err(|_| FixedBrokerProcessFailure::Protocol)?
{
Ok(reported) => reported,
Err(BrokerGateExit::ServiceGoneBeforeActivation | BrokerGateExit::ServiceGone) => {
return Err(FixedBrokerProcessFailure::ServiceGone);
}
};
let committed = match reported
.wait_for_ready_commit()
.map_err(|_| FixedBrokerProcessFailure::Protocol)?
{
Ok(committed) => committed,
Err(BrokerGateExit::ServiceGoneBeforeActivation | BrokerGateExit::ServiceGone) => {
return Err(FixedBrokerProcessFailure::ServiceGone);
}
};
committed
.resume_target()
.map_err(|_| FixedBrokerProcessFailure::Launcher)?
.wait_for_exit()
.map_err(|_| FixedBrokerProcessFailure::Launcher)?;
Ok(())
}
pub(in crate::backend::macos) unsafe fn run_fixed_gate_process(installed_path: &CStr) -> ! {
let adopted = unsafe { DormantBrokerGate::adopt_fixed_process(installed_path) };
let status = match adopted {
Err(_) => 64,
Ok(dormant) => match dormant.stage_plan() {
Ok(Err(BrokerGateExit::ServiceGoneBeforeActivation | BrokerGateExit::ServiceGone)) => 0,
Err(_) => 65,
Ok(Ok(staged)) => match staged.wait_for_activation() {
Ok(Err(
BrokerGateExit::ServiceGoneBeforeActivation | BrokerGateExit::ServiceGone,
)) => 0,
Err(_) => 65,
Ok(Ok(active)) => {
let ActiveBrokerProcess { gate, plan, trace } = active;
let _plan = plan;
drop(trace);
match gate.wait_for_service_death() {
Ok(BrokerGateExit::ServiceGone) => 0,
Ok(BrokerGateExit::ServiceGoneBeforeActivation) => 66,
Err(_) => 65,
}
}
},
},
};
unsafe { _exit(status) }
}
fn validate_fixed_arguments(
installed_path: &CStr,
arguments: impl IntoIterator<Item = impl AsRef<OsStr>>,
) -> Result<(), BrokerEntryError> {
if !is_deployer_helper_path(installed_path) {
return Err(BrokerEntryError::InvalidArguments);
}
let mut arguments = arguments.into_iter();
let expected = [
installed_path.to_bytes(),
INSTALLED_BROKER_MODE.as_bytes(),
INSTALLED_GATE_ARGUMENT.as_bytes(),
INSTALLED_CONTROL_ARGUMENT.as_bytes(),
INSTALLED_TRACE_ARGUMENT.as_bytes(),
];
for expected in expected {
let Some(argument) = arguments.next() else {
return Err(BrokerEntryError::InvalidArguments);
};
if argument.as_ref().as_bytes() != expected {
return Err(BrokerEntryError::InvalidArguments);
}
}
if arguments.next().is_some() {
return Err(BrokerEntryError::InvalidArguments);
}
Ok(())
}
fn read_retry(fd: c_int, byte: &mut u8) -> Result<isize, BrokerEntryError> {
loop {
match read_once(fd, byte) {
Err(error) if error == EINTR => continue,
Err(error) => return Err(BrokerEntryError::Read(error)),
Ok(count) => return Ok(count),
}
}
}
fn read_once(fd: c_int, byte: &mut u8) -> Result<isize, c_int> {
let count = unsafe { read(fd, byte, 1) };
if count < 0 {
Err(last_errno())
} else {
Ok(count)
}
}
fn set_nonblocking(fd: c_int, enabled: bool) -> Result<(), BrokerEntryError> {
let flags = unsafe { fcntl(fd, F_GETFL) };
if flags < 0 {
return Err(BrokerEntryError::Descriptor(last_errno()));
}
let desired = if enabled {
flags | O_NONBLOCK
} else {
flags & !O_NONBLOCK
};
if unsafe { fcntl(fd, F_SETFL, desired) } != 0 {
return Err(BrokerEntryError::Descriptor(last_errno()));
}
Ok(())
}
fn last_errno() -> c_int {
std::io::Error::last_os_error().raw_os_error().unwrap_or(0)
}
#[cfg(test)]
#[path = "supervisor_broker_entry_test.rs"]
mod tests;