ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Compose foreground sandbox process lifetimes

Browse source
This commit is contained in:
ms2k 2026-07-15 19:09:24 +09:00
commit fd78804fc7
15 changed files with 2313 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

7
crates/lazaret-host-coordinator/Cargo.toml
View file

@ -9,11 +9,17 @@ license.workspace = true
name = "lazaret-host-coordinator"
path = "src/main.rs"
[[test]]
name = "runtime-composition"
path = "tests/runtime-composition.rs"
harness = false
[dependencies]
ed25519-dalek.workspace = true
getrandom.workspace = true
lazaret-core = { version = "=0.0.0", path = "../lazaret-core" }
lazaret-linux = { version = "=0.0.0", path = "../lazaret-linux" }
lazaret-runtime-supervisor = { version = "=0.0.0", path = "../lazaret-runtime-supervisor" }
lazaret-state-anchor = { version = "=0.0.0", path = "../lazaret-state-anchor" }
lazaret-storage-catalog = { version = "=0.0.0", path = "../lazaret-storage-catalog" }
rcgen.workspace = true
@ -25,6 +31,7 @@ sha2.workspace = true
time.workspace = true
[dev-dependencies]
tempfile = "=3.27.0"
x509-parser.workspace = true
[lints]

1
crates/lazaret-host-coordinator/src/lib.rs
View file

@ -2,4 +2,5 @@
pub mod control;
pub mod domain_store;
pub mod runtime;
pub mod state_anchor;

9
crates/lazaret-host-coordinator/src/main.rs
View file

@ -7,5 +7,14 @@ fn main() {
}
return;
}
if let Some(result) =
lazaret_runtime_supervisor::supervision::dispatch_private_runtime_role(&arguments)
{
if let Err(error) = result {
eprintln!("Runtime Supervisor failed: {error}");
std::process::exit(1);
}
return;
}
let _ = lazaret_core::PROTOCOL_NAME;
}

690
crates/lazaret-host-coordinator/src/runtime.rs Normal file
View file

@ -0,0 +1,690 @@
use std::fmt;
use std::io;
use std::os::unix::ffi::OsStrExt;
use std::path::Path;
use std::process::Stdio;
use std::time::{Duration, Instant};
use lazaret_core::identity::{
BundleDigest, ContainerId, GenerationId, HostPath, PidfsProcessIdentityV1,
};
use lazaret_core::records::{
HostProcessOwnershipSlot, Participant, ParticipantOwnership, ParticipantRole, ParticipantState,
PersistedHostProcessOwnership, ProcessOwnershipCommitAck, SandboxOwnership,
StagedProcessOwnership,
};
use lazaret_linux::ipc::private_stream_pair;
use lazaret_linux::process::{
ChildStandardIo, RetainedProcess, child_broker, current_process_identity,
};
use lazaret_runtime_supervisor::supervision::{
AttachedRunscSignal, CoordinatorChannel, RuntimeSupervisorLaunch, SupervisorEvent,
};
use rustix::pipe::{PipeFlags, pipe_with};
use rustix::process::Signal;
use crate::domain_store::{DomainStoreError, DomainStoreMutation, HostCoordinatorDomainStore};
const SUPERVISOR_EXIT_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Debug)]
pub enum SandboxCoordinatorError {
Store(DomainStoreError),
Io(io::Error),
InvalidState(&'static str),
}
impl fmt::Display for SandboxCoordinatorError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Store(error) => {
write!(formatter, "Sandbox Domain Store operation failed: {error}")
}
Self::Io(error) => write!(formatter, "Sandbox process operation failed: {error}"),
Self::InvalidState(detail) => write!(formatter, "Sandbox state refused: {detail}"),
}
}
}
impl std::error::Error for SandboxCoordinatorError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Store(error) => Some(error),
Self::Io(error) => Some(error),
Self::InvalidState(_) => None,
}
}
}
impl From<DomainStoreError> for SandboxCoordinatorError {
fn from(error: DomainStoreError) -> Self {
Self::Store(error)
}
}
impl From<io::Error> for SandboxCoordinatorError {
fn from(error: io::Error) -> Self {
Self::Io(error)
}
}
pub struct SandboxCoordinator<'store> {
domain_store: &'store HostCoordinatorDomainStore,
}
impl<'store> SandboxCoordinator<'store> {
#[must_use]
pub const fn new(domain_store: &'store HostCoordinatorDomainStore) -> Self {
Self { domain_store }
}
/// Persists bare Sandbox intent and the immutable physical launch snapshot.
///
/// # Errors
/// Refuses a missing generation, changed physical identity, or existing nonmatching state.
pub fn begin_intent(
&self,
generation_id: GenerationId,
container_id: ContainerId,
runtime_root: &Path,
bundle_digest: BundleDigest,
host_coordinator: PidfsProcessIdentityV1,
) -> Result<(), SandboxCoordinatorError> {
let snapshot = self.domain_store.snapshot()?;
if !snapshot
.generations
.iter()
.any(|generation| generation.generation_id == generation_id)
{
return Err(SandboxCoordinatorError::InvalidState(
"Sandbox generation is absent",
));
}
let ownership = SandboxOwnership {
container_id: Some(container_id),
runtime_root: Some(HostPath::new(runtime_root.as_os_str().as_bytes().to_vec())),
bundle_digest: Some(bundle_digest),
host_coordinator: PersistedHostProcessOwnership::committed(host_coordinator),
runtime_supervisor: PersistedHostProcessOwnership::unassigned(),
attached_runsc: PersistedHostProcessOwnership::unassigned(),
};
let intent = sandbox_participant(generation_id, ParticipantState::Intent, ownership)?;
match sandbox_role(&snapshot.participants, generation_id) {
None => self.domain_store.apply(&[
DomainStoreMutation::PutParticipant(Participant::bare_intent(
generation_id,
ParticipantRole::Sandbox,
)),
DomainStoreMutation::PutParticipant(intent),
])?,
Some(current) if current == &intent => {}
Some(_) => {
return Err(SandboxCoordinatorError::InvalidState(
"Sandbox intent already differs",
));
}
}
Ok(())
}
/// # Errors
/// Refuses a mismatched RS tuple or any skipped ownership phase.
pub fn stage_runtime_supervisor(
&self,
staged: StagedProcessOwnership,
) -> Result<(), SandboxCoordinatorError> {
self.require_staged(staged, HostProcessOwnershipSlot::SandboxRuntimeSupervisor)?;
let (current, mut ownership) = self.intent_ownership(staged.generation_id())?;
let expected = PersistedHostProcessOwnership::staged(staged.identity());
if ownership.runtime_supervisor == expected {
return Ok(());
}
if !ownership.runtime_supervisor.is_unassigned() {
return Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor ownership changed",
));
}
ownership.runtime_supervisor = expected;
self.put_intent(current.generation_id(), ownership)
}
/// # Errors
/// Refuses an RS staged identity that is not durably exact.
pub fn acknowledge_runtime_supervisor(
&self,
staged: StagedProcessOwnership,
) -> Result<ProcessOwnershipCommitAck, SandboxCoordinatorError> {
self.require_staged(staged, HostProcessOwnershipSlot::SandboxRuntimeSupervisor)?;
let (_, ownership) = self.intent_ownership(staged.generation_id())?;
if ownership.runtime_supervisor != PersistedHostProcessOwnership::staged(staged.identity())
{
return Err(SandboxCoordinatorError::InvalidState(
"staged Runtime Supervisor ownership is not durable",
));
}
Ok(ProcessOwnershipCommitAck::v1(staged))
}
/// Commits the executing RS and durably stages its exact attached-runsc child.
///
/// # Errors
/// Refuses any tuple, phase, or immutable-evidence mismatch.
pub fn stage_attached_runsc(
&self,
staged: StagedProcessOwnership,
) -> Result<ProcessOwnershipCommitAck, SandboxCoordinatorError> {
self.require_staged(staged, HostProcessOwnershipSlot::SandboxAttachedRunsc)?;
let (current, mut ownership) = self.intent_ownership(staged.generation_id())?;
let rs = *ownership
.runtime_supervisor
.authoritative_pidfs()
.map_err(|_| {
SandboxCoordinatorError::InvalidState("Runtime Supervisor is not staged")
})?;
if !ownership.runtime_supervisor.is_staged() {
return Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor readiness cannot skip staging",
));
}
if !ownership.attached_runsc.is_unassigned() {
return Err(SandboxCoordinatorError::InvalidState(
"attached runsc ownership changed",
));
}
ownership.runtime_supervisor = PersistedHostProcessOwnership::committed(rs);
ownership.attached_runsc = PersistedHostProcessOwnership::staged(staged.identity());
self.put_intent(current.generation_id(), ownership)?;
Ok(ProcessOwnershipCommitAck::v1(staged))
}
/// Promotes the exact attached runsc only after the RS reports target readiness.
///
/// # Errors
/// Refuses any incomplete physical snapshot or changed staged identity.
pub fn commit_ready(
&self,
staged: StagedProcessOwnership,
) -> Result<(), SandboxCoordinatorError> {
self.require_staged(staged, HostProcessOwnershipSlot::SandboxAttachedRunsc)?;
let (_, mut ownership) = self.intent_ownership(staged.generation_id())?;
if ownership.attached_runsc != PersistedHostProcessOwnership::staged(staged.identity())
|| !ownership.runtime_supervisor.is_current_committed()
|| !ownership.host_coordinator.is_current_committed()
{
return Err(SandboxCoordinatorError::InvalidState(
"Sandbox readiness does not match staged ownership",
));
}
ownership.attached_runsc = PersistedHostProcessOwnership::committed(staged.identity());
self.domain_store
.apply(&[DomainStoreMutation::PutParticipant(sandbox_participant(
staged.generation_id(),
ParticipantState::Committed,
ownership,
)?)])?;
Ok(())
}
/// # Errors
/// Refuses any Sandbox participant that is not exactly Committed.
pub fn begin_release(
&self,
generation_id: GenerationId,
) -> Result<(), SandboxCoordinatorError> {
let current = self.participant(generation_id)?;
if current.state() != ParticipantState::Committed {
return Err(SandboxCoordinatorError::InvalidState(
"Sandbox is not committed",
));
}
let ownership =
current
.ownership()
.cloned()
.ok_or(SandboxCoordinatorError::InvalidState(
"Sandbox ownership is absent",
))?;
self.domain_store
.apply(&[DomainStoreMutation::PutParticipant(
Participant::snapshot(
generation_id,
ParticipantRole::Sandbox,
ParticipantState::ReleaseIntent,
Some(ownership),
)
.map_err(|_| SandboxCoordinatorError::InvalidState("Sandbox release is invalid"))?,
)])?;
Ok(())
}
/// # Errors
/// Refuses cleanup without exact `ReleaseIntent` and observed runtime absence.
pub fn complete_clean_after_absence(
&self,
generation_id: GenerationId,
) -> Result<(), SandboxCoordinatorError> {
let current = self.participant(generation_id)?;
if current.state() != ParticipantState::ReleaseIntent {
return Err(SandboxCoordinatorError::InvalidState(
"Sandbox release intent is absent",
));
}
let clean = Participant::snapshot(
generation_id,
ParticipantRole::Sandbox,
ParticipantState::Clean,
Some(ParticipantOwnership::empty(ParticipantRole::Sandbox)),
)
.map_err(|_| SandboxCoordinatorError::InvalidState("Sandbox Clean is invalid"))?;
self.domain_store
.apply(&[DomainStoreMutation::CleanParticipantAfterAbsence(clean)])?;
Ok(())
}
fn require_staged(
&self,
staged: StagedProcessOwnership,
slot: HostProcessOwnershipSlot,
) -> Result<(), SandboxCoordinatorError> {
let snapshot = self.domain_store.snapshot()?;
let generation = snapshot
.generations
.iter()
.find(|generation| generation.generation_id == staged.generation_id())
.ok_or(SandboxCoordinatorError::InvalidState(
"staged Sandbox generation is absent",
))?;
if staged.domain_id() != snapshot.domain.domain_id
|| staged.coordinator_id() != generation.coordinator_id
|| staged.slot() != slot
{
return Err(SandboxCoordinatorError::InvalidState(
"staged Sandbox ownership tuple mismatches generation",
));
}
Ok(())
}
fn intent_ownership(
&self,
generation_id: GenerationId,
) -> Result<(Participant, SandboxOwnership), SandboxCoordinatorError> {
let current = self.participant(generation_id)?;
if current.state() != ParticipantState::Intent {
return Err(SandboxCoordinatorError::InvalidState(
"Sandbox acquisition is not in Intent",
));
}
let Some(ParticipantOwnership::Sandbox(ownership)) = current.ownership() else {
return Err(SandboxCoordinatorError::InvalidState(
"Sandbox ownership shape is invalid",
));
};
Ok((current.clone(), ownership.clone()))
}
fn put_intent(
&self,
generation_id: GenerationId,
ownership: SandboxOwnership,
) -> Result<(), SandboxCoordinatorError> {
self.domain_store
.apply(&[DomainStoreMutation::PutParticipant(sandbox_participant(
generation_id,
ParticipantState::Intent,
ownership,
)?)])?;
Ok(())
}
fn participant(
&self,
generation_id: GenerationId,
) -> Result<Participant, SandboxCoordinatorError> {
sandbox_role(&self.domain_store.snapshot()?.participants, generation_id)
.cloned()
.ok_or(SandboxCoordinatorError::InvalidState(
"Sandbox participant is absent",
))
}
}
pub struct AwaitingAttachedRunsc {
supervisor: Option<RetainedProcess>,
control: Option<CoordinatorChannel>,
lifetime: Option<std::os::fd::OwnedFd>,
staged: Option<StagedProcessOwnership>,
}
impl AwaitingAttachedRunsc {
/// Launches the RS through the HC broker and commits its staged ownership before execution.
///
/// # Errors
/// Refuses any pipe, channel, broker, staging, or acknowledgment failure.
pub fn launch(
coordinator: &SandboxCoordinator<'_>,
launch: &RuntimeSupervisorLaunch,
) -> Result<Self, SandboxCoordinatorError> {
let parent = current_process_identity()?;
let (coordinator_control, supervisor_control) = private_stream_pair()?;
let (supervisor_lifetime, coordinator_lifetime) =
pipe_with(PipeFlags::CLOEXEC).map_err(io::Error::from)?;
let process = child_broker().acquire_with_stdio(
parent,
launch.ownership(),
launch.target(),
ChildStandardIo::redirected(
Stdio::from(supervisor_control),
Stdio::from(supervisor_lifetime),
),
|staged| {
coordinator
.stage_runtime_supervisor(staged)
.map_err(io::Error::other)
},
|staged| {
coordinator
.acknowledge_runtime_supervisor(staged)
.map_err(io::Error::other)
},
)?;
Ok(Self {
supervisor: Some(process),
control: Some(CoordinatorChannel::new(coordinator_control)),
lifetime: Some(coordinator_lifetime),
staged: None,
})
}
/// Receives and durably stages the exact RS child without releasing its barrier.
///
/// # Errors
/// Refuses unexpected, duplicate, or mismatched messages and durable-state failures.
pub fn receive_staged(
&mut self,
coordinator: &SandboxCoordinator<'_>,
) -> Result<StagedProcessOwnership, SandboxCoordinatorError> {
if self.staged.is_some() {
return Err(SandboxCoordinatorError::InvalidState(
"attached runsc stage was already received",
));
}
let event = self
.control
.as_ref()
.ok_or(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor control channel is absent",
))?
.receive()?;
let staged = match event {
SupervisorEvent::Staged(staged) => staged,
SupervisorEvent::Refused(detail) => {
return Err(SandboxCoordinatorError::Io(io::Error::other(detail)));
}
_ => {
return Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor sent readiness before staged ownership",
));
}
};
let acknowledgment = coordinator.stage_attached_runsc(staged)?;
self.staged = Some(acknowledgment.committed());
Ok(staged)
}
/// Releases the exact pending runsc barrier with its durable typed acknowledgment.
///
/// # Errors
/// Refuses a different or absent staged tuple or channel failure.
pub fn acknowledge(
mut self,
staged: StagedProcessOwnership,
) -> Result<AwaitingSandboxReadiness, SandboxCoordinatorError> {
if self.staged != Some(staged) {
return Err(SandboxCoordinatorError::InvalidState(
"attached runsc acknowledgment does not match pending stage",
));
}
self.control
.as_ref()
.ok_or(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor control channel is absent",
))?
.acknowledge_staged(ProcessOwnershipCommitAck::v1(staged))?;
Ok(AwaitingSandboxReadiness {
supervisor: self.supervisor.take(),
control: self.control.take(),
lifetime: self.lifetime.take(),
staged,
})
}
/// # Errors
/// Returns an error if the exact RS cannot be terminated and reaped.
pub fn abort(mut self) -> Result<(), SandboxCoordinatorError> {
self.lifetime.take();
cleanup_supervisor(&mut self.supervisor)?;
Ok(())
}
}
impl Drop for AwaitingAttachedRunsc {
fn drop(&mut self) {
self.lifetime.take();
let _ = cleanup_supervisor(&mut self.supervisor);
}
}
pub struct AwaitingSandboxReadiness {
supervisor: Option<RetainedProcess>,
control: Option<CoordinatorChannel>,
lifetime: Option<std::os::fd::OwnedFd>,
staged: StagedProcessOwnership,
}
impl AwaitingSandboxReadiness {
/// Commits only the exact ready attached runsc after RS capability removal.
///
/// # Errors
/// Refuses a mismatched readiness tuple, durable commit failure, or channel failure.
pub fn commit(
mut self,
coordinator: &SandboxCoordinator<'_>,
) -> Result<CommittedSandboxRuntime, SandboxCoordinatorError> {
let control = self
.control
.as_ref()
.ok_or(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor control channel is absent",
))?;
let event = control.receive()?;
let staged = match event {
SupervisorEvent::Ready { staged, .. } => staged,
SupervisorEvent::Refused(detail) => {
return Err(SandboxCoordinatorError::Io(io::Error::other(detail)));
}
_ => {
return Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor did not report sandbox readiness",
));
}
};
if staged != self.staged {
return Err(SandboxCoordinatorError::InvalidState(
"sandbox readiness changed attached-runsc ownership",
));
}
coordinator.commit_ready(staged)?;
control.acknowledge_sandbox_committed()?;
Ok(CommittedSandboxRuntime {
generation_id: staged.generation_id(),
supervisor: self.supervisor.take(),
control: self
.control
.take()
.ok_or(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor control channel is absent",
))?,
lifetime: self.lifetime.take(),
})
}
}
impl Drop for AwaitingSandboxReadiness {
fn drop(&mut self) {
self.lifetime.take();
let _ = cleanup_supervisor(&mut self.supervisor);
}
}
pub struct CommittedSandboxRuntime {
generation_id: GenerationId,
supervisor: Option<RetainedProcess>,
control: CoordinatorChannel,
lifetime: Option<std::os::fd::OwnedFd>,
}
impl CommittedSandboxRuntime {
#[must_use]
pub fn runtime_supervisor_identity(&self) -> Option<PidfsProcessIdentityV1> {
self.supervisor.as_ref().map(RetainedProcess::identity)
}
/// # Errors
/// Returns an error if the exact retained RS pidfd action fails.
pub fn signal_runtime_supervisor(&self, signal: Signal) -> Result<(), SandboxCoordinatorError> {
self.supervisor
.as_ref()
.ok_or(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor handle is absent",
))?
.send_signal(signal)?;
Ok(())
}
/// # Errors
/// Returns an error if the RS cannot apply the exact retained-runsc pidfd action.
pub fn signal_attached_runsc(
&self,
signal: AttachedRunscSignal,
) -> Result<(), SandboxCoordinatorError> {
self.control.signal_attached_runsc(signal)?;
if self.control.receive()? == SupervisorEvent::SignalApplied(signal) {
Ok(())
} else {
Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor did not confirm the exact runsc action",
))
}
}
/// Waits for an exact attached-runsc loss report and reaps the retained RS.
///
/// # Errors
/// Refuses any other event or an RS that does not then exit.
pub fn wait_for_attached_runsc_exit(mut self) -> Result<(), SandboxCoordinatorError> {
if self.control.receive()? != SupervisorEvent::AttachedRunscExited {
return Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor did not report attached-runsc loss",
));
}
self.lifetime.take();
wait_and_reap_supervisor(&mut self.supervisor)
}
/// Waits for an externally stopped exact RS handle to exit and reaps it.
///
/// # Errors
/// Refuses an RS that remains present beyond the bounded observation.
pub fn wait_for_runtime_supervisor_exit(mut self) -> Result<(), SandboxCoordinatorError> {
self.lifetime.take();
wait_and_reap_supervisor(&mut self.supervisor)
}
/// Performs ordered normal stop and reaps the same retained RS handle.
///
/// # Errors
/// Refuses missing release intent, unexpected RS response, or exact cleanup failure.
pub fn stop(
mut self,
coordinator: &SandboxCoordinator<'_>,
) -> Result<(), SandboxCoordinatorError> {
coordinator.begin_release(self.generation_id)?;
self.control.stop()?;
if self.control.receive()? != SupervisorEvent::Stopped {
return Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor did not complete sandbox stop",
));
}
self.lifetime.take();
wait_and_reap_supervisor(&mut self.supervisor)?;
coordinator.complete_clean_after_absence(self.generation_id)
}
}
impl Drop for CommittedSandboxRuntime {
fn drop(&mut self) {
self.lifetime.take();
let _ = cleanup_supervisor(&mut self.supervisor);
}
}
fn sandbox_participant(
generation_id: GenerationId,
state: ParticipantState,
ownership: SandboxOwnership,
) -> Result<Participant, SandboxCoordinatorError> {
Participant::snapshot(
generation_id,
ParticipantRole::Sandbox,
state,
Some(ParticipantOwnership::Sandbox(ownership)),
)
.map_err(|_| SandboxCoordinatorError::InvalidState("Sandbox participant shape is invalid"))
}
fn sandbox_role(participants: &[Participant], generation_id: GenerationId) -> Option<&Participant> {
participants.iter().find(|participant| {
participant.generation_id() == generation_id
&& participant.role() == ParticipantRole::Sandbox
})
}
fn cleanup_supervisor(
supervisor: &mut Option<RetainedProcess>,
) -> Result<(), SandboxCoordinatorError> {
let Some(process) = supervisor.take() else {
return Ok(());
};
if !process.has_exited()? {
process.send_signal(Signal::KILL)?;
}
child_broker().reap(process)?;
Ok(())
}
fn wait_and_reap_supervisor(
supervisor: &mut Option<RetainedProcess>,
) -> Result<(), SandboxCoordinatorError> {
let Some(process) = supervisor.take() else {
return Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor handle is absent",
));
};
let deadline = Instant::now() + SUPERVISOR_EXIT_TIMEOUT;
while !process.has_exited()? && Instant::now() < deadline {
std::thread::sleep(Duration::from_millis(5));
}
if !process.has_exited()? {
process.send_signal(Signal::KILL)?;
child_broker().reap(process)?;
return Err(SandboxCoordinatorError::InvalidState(
"Runtime Supervisor did not exit after sandbox stop",
));
}
child_broker().reap(process)?;
Ok(())
}
#[cfg(test)]
mod tests;

179
crates/lazaret-host-coordinator/src/runtime/tests.rs Normal file
View file

@ -0,0 +1,179 @@
use std::fs::{File, OpenOptions};
use std::path::Path;
use lazaret_core::identity::{
BootId, ConfigurationFingerprint, CoordinatorId, DomainId, HostPidLocator, NetworkSidecarId,
PidfsDeviceId, PidfsInodeId, PidfsProcessAuthorityV1, ProcessStartTimeTicks, SandboxId,
};
use lazaret_core::records::{Domain, DurableDecision, Generation};
use lazaret_linux::filesystem::try_exclusive_flock;
use tempfile::TempDir;
use super::*;
fn bytes(value: u8) -> [u8; 32] {
[value; 32]
}
fn process(value: u32) -> PidfsProcessIdentityV1 {
PidfsProcessIdentityV1::new(
PidfsProcessAuthorityV1 {
boot_id: BootId::new([1; 16]),
pidfs_st_dev: PidfsDeviceId::new(u64::from(value)),
pidfs_st_ino: PidfsInodeId::new(u64::from(value) + 100),
},
HostPidLocator::new(value),
ProcessStartTimeTicks::new(u64::from(value) + 200),
)
}
fn domain() -> Domain {
Domain {
domain_id: DomainId::new(bytes(1)),
profile_name: "trusted".into(),
canonical_workspace: HostPath::new(b"/workspace/project".to_vec()),
}
}
fn generation() -> Generation {
Generation {
domain_id: domain().domain_id,
generation_id: GenerationId::new(7),
configuration_fingerprint: ConfigurationFingerprint::new(bytes(2)),
coordinator_id: CoordinatorId::new(bytes(3)),
network_sidecar_id: NetworkSidecarId::new(bytes(4)),
sandbox_id: SandboxId::new(bytes(5)),
durable_decision: DurableDecision::None,
}
}
fn lock_file(path: &Path) -> File {
OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(path)
.unwrap()
}
fn store(directory: &TempDir) -> HostCoordinatorDomainStore {
let lock = try_exclusive_flock(lock_file(&directory.path().join("domain.lock"))).unwrap();
let store =
HostCoordinatorDomainStore::open(lock, &directory.path().join("domain.redb"), &domain())
.unwrap();
store
.apply(&[DomainStoreMutation::PutGeneration(generation())])
.unwrap();
store
}
fn staged(
slot: HostProcessOwnershipSlot,
identity: PidfsProcessIdentityV1,
) -> StagedProcessOwnership {
StagedProcessOwnership::v1(
domain().domain_id,
generation().generation_id,
generation().coordinator_id,
slot,
identity,
)
}
#[test]
fn sandbox_staging_readiness_and_release_preserve_exact_ownership() {
let directory = tempfile::tempdir().unwrap();
let store = store(&directory);
let coordinator = SandboxCoordinator::new(&store);
coordinator
.begin_intent(
generation().generation_id,
ContainerId::new(bytes(6)),
directory.path(),
BundleDigest::new(bytes(7)),
process(100),
)
.unwrap();
let rs = staged(
HostProcessOwnershipSlot::SandboxRuntimeSupervisor,
process(101),
);
coordinator.stage_runtime_supervisor(rs).unwrap();
assert!(
coordinator
.acknowledge_runtime_supervisor(rs)
.unwrap()
.acknowledges(&rs)
);
let runsc = staged(HostProcessOwnershipSlot::SandboxAttachedRunsc, process(102));
assert!(
coordinator
.stage_attached_runsc(runsc)
.unwrap()
.acknowledges(&runsc)
);
let intent = coordinator.participant(generation().generation_id).unwrap();
let Some(ParticipantOwnership::Sandbox(ownership)) = intent.ownership() else {
panic!("Sandbox ownership missing");
};
assert_eq!(intent.state(), ParticipantState::Intent);
assert_eq!(
ownership.runtime_supervisor,
PersistedHostProcessOwnership::committed(process(101))
);
assert_eq!(
ownership.attached_runsc,
PersistedHostProcessOwnership::staged(process(102))
);
coordinator.commit_ready(runsc).unwrap();
assert_eq!(
coordinator
.participant(generation().generation_id)
.unwrap()
.state(),
ParticipantState::Committed
);
coordinator
.begin_release(generation().generation_id)
.unwrap();
coordinator
.complete_clean_after_absence(generation().generation_id)
.unwrap();
let clean = coordinator.participant(generation().generation_id).unwrap();
assert_eq!(clean.state(), ParticipantState::Clean);
assert!(clean.ownership().unwrap().is_empty());
}
#[test]
fn mismatched_and_skipped_acquisition_boundaries_refuse_without_state_change() {
let directory = tempfile::tempdir().unwrap();
let store = store(&directory);
let coordinator = SandboxCoordinator::new(&store);
coordinator
.begin_intent(
generation().generation_id,
ContainerId::new(bytes(6)),
directory.path(),
BundleDigest::new(bytes(7)),
process(100),
)
.unwrap();
let runsc = staged(HostProcessOwnershipSlot::SandboxAttachedRunsc, process(102));
assert!(coordinator.stage_attached_runsc(runsc).is_err());
assert!(coordinator.commit_ready(runsc).is_err());
let wrong_domain = StagedProcessOwnership::v1(
DomainId::new(bytes(9)),
generation().generation_id,
generation().coordinator_id,
HostProcessOwnershipSlot::SandboxRuntimeSupervisor,
process(101),
);
assert!(coordinator.stage_runtime_supervisor(wrong_domain).is_err());
let participant = coordinator.participant(generation().generation_id).unwrap();
let Some(ParticipantOwnership::Sandbox(ownership)) = participant.ownership() else {
panic!("Sandbox ownership missing");
};
assert!(ownership.runtime_supervisor.is_unassigned());
assert!(ownership.attached_runsc.is_unassigned());
}

558
crates/lazaret-host-coordinator/tests/runtime-composition.rs Normal file
View file

@ -0,0 +1,558 @@
use std::collections::BTreeMap;
use std::fs::{self, File, OpenOptions};
use std::os::unix::ffi::OsStringExt;
use std::os::unix::net::UnixListener;
use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::thread;
use std::time::Duration;
use lazaret_core::configuration::{
ConfigurationLoadRequest, ResolvedDomainConfiguration, load_trusted_configuration,
};
use lazaret_core::identity::{
BundleDigest, ContainerId, CoordinatorId, GenerationId, HostPath, NetworkSidecarId, SandboxId,
};
use lazaret_core::records::{
Domain, DurableDecision, Generation, HostProcessOwnershipSlot, ParticipantRole,
ParticipantState, ProcessOwnershipCommitAck,
};
use lazaret_host_coordinator::domain_store::{DomainStoreMutation, HostCoordinatorDomainStore};
use lazaret_host_coordinator::runtime::{
AwaitingAttachedRunsc, CommittedSandboxRuntime, SandboxCoordinator,
};
use lazaret_linux::filesystem::try_exclusive_flock;
use lazaret_linux::platform::{PlatformObservation, check_platform};
use lazaret_linux::process::{
ProcessOwnershipContext, RetainedProcess, TargetCommand, child_broker, current_process_identity,
};
use lazaret_runtime_supervisor::sandbox::{
NETWORK_NAMESPACE_PATH, RootProcess, SandboxBuildRequest, SandboxBundle, build_sandbox_bundle,
production_sandbox_layout,
};
use lazaret_runtime_supervisor::supervision::{
AttachedRunscSignal, RuntimeSupervisorLaunch, dispatch_private_runtime_role,
};
use rustix::process::Signal;
use tempfile::TempDir;
const SKIP: u8 = 77;
const HOST_COORDINATOR_DRIVER: &str = "--runtime-composition-host-coordinator";
fn main() -> ExitCode {
let arguments = std::env::args_os().skip(1).collect::<Vec<_>>();
if let Some(result) = lazaret_linux::process::dispatch_private_process_role(&arguments) {
return exit(result);
}
if let Some(result) = dispatch_private_runtime_role(&arguments) {
return exit(result);
}
if arguments.first().and_then(|argument| argument.to_str()) == Some(HOST_COORDINATOR_DRIVER) {
return exit(run_host_coordinator_driver(&arguments[1..]));
}
match run_matrix() {
Ok(()) => {
println!("PASS foreground runtime composition and causal-loss matrix");
ExitCode::SUCCESS
}
Err(MatrixError::Skip(reason)) => {
println!("SKIP foreground runtime composition: {reason}");
ExitCode::from(SKIP)
}
Err(MatrixError::Failure(error)) => {
eprintln!("runtime composition failed: {error}");
ExitCode::FAILURE
}
}
}
fn exit(result: std::io::Result<()>) -> ExitCode {
match result {
Ok(()) => ExitCode::SUCCESS,
Err(error) => {
eprintln!("private runtime role failed: {error}");
ExitCode::FAILURE
}
}
}
enum MatrixError {
Skip(String),
Failure(String),
}
impl From<std::io::Error> for MatrixError {
fn from(error: std::io::Error) -> Self {
Self::Failure(error.to_string())
}
}
fn run_matrix() -> Result<(), MatrixError> {
let release = fs::read_to_string("/proc/sys/kernel/osrelease")?;
check_platform(&PlatformObservation {
operating_system: std::env::consts::OS,
architecture: std::env::consts::ARCH,
kernel_release: release.trim(),
})
.map_err(|error| MatrixError::Skip(error.to_string()))?;
require_runtime_prerequisites()?;
lost_acknowledgment_refuses_target_work()?;
normal_stop_and_stop_resume()?;
attached_runsc_loss_stops_guest()?;
runtime_supervisor_loss_stops_guest()?;
host_coordinator_loss_stops_guest()
}
fn require_runtime_prerequisites() -> Result<(), MatrixError> {
let runsc = std::env::var_os("PATH")
.into_iter()
.flat_map(|path| std::env::split_paths(&path).collect::<Vec<_>>())
.map(|directory| directory.join("runsc"))
.find(|candidate| candidate.is_file())
.ok_or_else(|| MatrixError::Skip("runsc is not installed on PATH".into()))?;
for path in [
runsc,
PathBuf::from("/bin/sh"),
PathBuf::from("/bin/sleep"),
PathBuf::from("/lib64/libc.so.6"),
PathBuf::from("/lib64/libtinfo.so.6"),
PathBuf::from("/lib64/ld-linux-x86-64.so.2"),
] {
if !path.is_file() {
return Err(MatrixError::Skip(format!(
"required runtime fixture is absent: {}",
path.display()
)));
}
}
let socket_directory = tempfile::tempdir()?;
let socket_path = socket_directory.path().join("runsc-control.sock");
match UnixListener::bind(&socket_path) {
Ok(listener) => drop(listener),
Err(error) if error.kind() == std::io::ErrorKind::PermissionDenied => {
return Err(MatrixError::Skip(format!(
"filesystem Unix sockets required by runsc are denied: {error}"
)));
}
Err(error) => return Err(error.into()),
}
Ok(())
}
fn lost_acknowledgment_refuses_target_work() -> Result<(), MatrixError> {
let fixture = Fixture::new()?;
let mut pending = fixture.launch_pending()?;
let staged = pending
.receive_staged(&SandboxCoordinator::new(&fixture.store))
.map_err(failure)?;
let participant = sandbox_participant(&fixture.store);
assert_eq!(participant.state(), ParticipantState::Intent);
let lazaret_core::records::ParticipantOwnership::Sandbox(ownership) =
participant.ownership().unwrap()
else {
unreachable!();
};
assert_eq!(
ownership.attached_runsc.authoritative_pidfs().unwrap(),
&staged.identity()
);
pending.abort().map_err(failure)?;
if fixture.activity.exists() {
return Err(failure("lost acknowledgment released sandbox work"));
}
Ok(())
}
fn normal_stop_and_stop_resume() -> Result<(), MatrixError> {
let fixture = Fixture::new()?;
let runtime = fixture.launch_committed()?;
wait_for_activity(&fixture.activity)?;
runtime
.signal_attached_runsc(AttachedRunscSignal::Stop)
.map_err(failure)?;
thread::sleep(Duration::from_millis(50));
runtime
.signal_attached_runsc(AttachedRunscSignal::Continue)
.map_err(failure)?;
runtime
.signal_runtime_supervisor(Signal::STOP)
.map_err(failure)?;
let stopped_size = fs::metadata(&fixture.activity)?.len();
thread::sleep(Duration::from_millis(50));
if fs::metadata(&fixture.activity)?.len() <= stopped_size {
return Err(failure(
"stopped Runtime Supervisor incorrectly stopped a present sandbox",
));
}
runtime
.signal_runtime_supervisor(Signal::CONT)
.map_err(failure)?;
runtime
.stop(&SandboxCoordinator::new(&fixture.store))
.map_err(failure)?;
assert_eq!(
sandbox_participant(&fixture.store).state(),
ParticipantState::Clean
);
assert_activity_stable(&fixture.activity)
}
fn attached_runsc_loss_stops_guest() -> Result<(), MatrixError> {
let fixture = Fixture::new()?;
let runtime = fixture.launch_committed()?;
wait_for_activity(&fixture.activity)?;
runtime
.signal_attached_runsc(AttachedRunscSignal::Kill)
.map_err(failure)?;
runtime.wait_for_attached_runsc_exit().map_err(failure)?;
assert_eq!(
sandbox_participant(&fixture.store).state(),
ParticipantState::Committed
);
assert_activity_stable(&fixture.activity)
}
fn runtime_supervisor_loss_stops_guest() -> Result<(), MatrixError> {
let fixture = Fixture::new()?;
let runtime = fixture.launch_committed()?;
wait_for_activity(&fixture.activity)?;
runtime
.signal_runtime_supervisor(Signal::KILL)
.map_err(failure)?;
runtime
.wait_for_runtime_supervisor_exit()
.map_err(failure)?;
assert_eq!(
sandbox_participant(&fixture.store).state(),
ParticipantState::Committed
);
assert_activity_stable(&fixture.activity)
}
fn host_coordinator_loss_stops_guest() -> Result<(), MatrixError> {
let tree = tempfile::tempdir()?;
let ready = tree.path().join("host-coordinator-ready");
let activity = tree.path().join("workspace/activity");
let parent = current_process_identity()?;
let target = TargetCommand::new(std::env::current_exe()?)
.argument(HOST_COORDINATOR_DRIVER)
.argument(tree.path())
.argument(&ready);
let mut driver = ExactChild::new(child_broker().acquire(
parent,
ProcessOwnershipContext {
domain_id: lazaret_core::identity::DomainId::new([1; 32]),
generation_id: GenerationId::new(7),
coordinator_id: CoordinatorId::new([3; 32]),
slot: HostProcessOwnershipSlot::SandboxHostCoordinator,
},
&target,
|_| Ok(()),
|staged| Ok(ProcessOwnershipCommitAck::v1(staged)),
)?);
for _ in 0..400 {
if ready.is_file() {
break;
}
if driver.process()?.has_exited()? {
driver.reap()?;
return Err(failure("Host Coordinator driver exited before readiness"));
}
thread::sleep(Duration::from_millis(5));
}
if !ready.is_file() {
driver.kill_and_reap()?;
return Err(failure("Host Coordinator driver readiness timed out"));
}
wait_for_activity(&activity)?;
driver.kill_and_reap()?;
thread::sleep(Duration::from_millis(100));
assert_activity_stable(&activity)
}
struct ExactChild(Option<RetainedProcess>);
impl ExactChild {
const fn new(process: RetainedProcess) -> Self {
Self(Some(process))
}
fn process(&self) -> std::io::Result<&RetainedProcess> {
self.0
.as_ref()
.ok_or_else(|| std::io::Error::other("exact test child handle is absent"))
}
fn reap(&mut self) -> std::io::Result<()> {
let process = self
.0
.take()
.ok_or_else(|| std::io::Error::other("exact test child handle is absent"))?;
child_broker().reap(process)
}
fn kill_and_reap(&mut self) -> std::io::Result<()> {
self.process()?.send_signal(Signal::KILL)?;
self.reap()
}
}
impl Drop for ExactChild {
fn drop(&mut self) {
if let Some(process) = self.0.take() {
let _ = process.send_signal(Signal::KILL);
let _ = child_broker().reap(process);
}
}
}
fn run_host_coordinator_driver(arguments: &[std::ffi::OsString]) -> std::io::Result<()> {
let tree = arguments
.first()
.map(PathBuf::from)
.ok_or_else(|| std::io::Error::other("Host Coordinator fixture path is missing"))?;
let ready = arguments
.get(1)
.map(PathBuf::from)
.ok_or_else(|| std::io::Error::other("Host Coordinator ready path is missing"))?;
if arguments.len() != 2 {
return Err(std::io::Error::other(
"Host Coordinator driver arguments are malformed",
));
}
let fixture = Fixture::in_directory(&tree).map_err(matrix_io)?;
let _runtime = fixture.launch_committed().map_err(matrix_io)?;
wait_for_activity(&fixture.activity).map_err(matrix_io)?;
fs::write(ready, b"ready")?;
loop {
thread::park();
}
}
fn matrix_io(error: MatrixError) -> std::io::Error {
match error {
MatrixError::Skip(detail) | MatrixError::Failure(detail) => std::io::Error::other(detail),
}
}
fn failure(error: impl std::fmt::Display) -> MatrixError {
MatrixError::Failure(error.to_string())
}
fn wait_for_activity(path: &Path) -> Result<(), MatrixError> {
for _ in 0..400 {
if path.metadata().is_ok_and(|metadata| metadata.len() >= 2) {
return Ok(());
}
thread::sleep(Duration::from_millis(5));
}
Err(failure("guest activity did not start"))
}
fn assert_activity_stable(path: &Path) -> Result<(), MatrixError> {
let before = fs::metadata(path)?.len();
thread::sleep(Duration::from_millis(100));
let after = fs::metadata(path)?.len();
if before == after {
Ok(())
} else {
Err(failure(
"guest activity continued after causal authority loss",
))
}
}
struct Fixture {
_tree: Option<TempDir>,
configuration: ResolvedDomainConfiguration,
store: HostCoordinatorDomainStore,
bundle: SandboxBundle,
activity: PathBuf,
}
impl Fixture {
fn new() -> Result<Self, MatrixError> {
let tree = tempfile::tempdir()?;
let root = tree.path().to_path_buf();
Self::build(&root, Some(tree))
}
fn in_directory(root: &Path) -> Result<Self, MatrixError> {
Self::build(root, None)
}
fn build(root: &Path, tree: Option<TempDir>) -> Result<Self, MatrixError> {
let configuration_directory = root.join("configuration");
let workspace = root.join("workspace");
let runtime = root.join("runtime");
fs::create_dir(&configuration_directory)?;
fs::create_dir(&workspace)?;
fs::create_dir(&runtime)?;
let configuration_path = configuration_directory.join("lazaret.toml");
fs::write(
&configuration_path,
format!(
"schema = \"lazaret-trusted-host-config-v1\"\n\n[storage]\ndata_root = \"{}\"\n\n[profiles.default]\nimage = \"example.invalid/runtime-fixture:v1\"\npersistent_dirs = []\n",
root.join("data").display()
),
)?;
let configuration = load_trusted_configuration(ConfigurationLoadRequest {
explicit_configuration_path: Some(configuration_path),
selected_profile: "default".into(),
workspace: workspace.clone(),
home: root.join("home"),
xdg_config_home: None,
xdg_data_home: None,
xdg_runtime_dir: runtime,
environment: BTreeMap::new(),
sandbox_layout: production_sandbox_layout().map_err(failure)?,
})
.map_err(failure)?;
let runtime_domain = host_path(configuration.runtime_domain_root().canonical_label());
stage_rootfs_and_control(&runtime_domain)?;
let container_id = ContainerId::new([6; 32]);
let root_process = RootProcess::new(
"/bin/sh",
[
"-c".to_owned(),
"while :; do echo x >> /workspace/activity; /bin/sleep 0.02; done".to_owned(),
],
)
.map_err(failure)?;
let namespace =
lazaret_linux::filesystem::namespace_identity(Path::new(NETWORK_NAMESPACE_PATH))?;
let bundle = build_sandbox_bundle(SandboxBuildRequest {
configuration: &configuration,
container_id,
expected_network_namespace: namespace,
root_process,
terminal: None,
})
.map_err(failure)?;
let bundle_path = runtime_domain
.join("bundles")
.join(hex(container_id.as_bytes()));
fs::create_dir_all(&bundle_path)?;
fs::write(bundle_path.join("config.json"), bundle.specification_json())?;
fs::create_dir_all(runtime_domain.join("runsc"))?;
let domain = Domain {
domain_id: configuration.domain_id(),
profile_name: configuration.selected_profile().to_owned(),
canonical_workspace: configuration.workspace().canonical_label().clone(),
};
let persistent_domain = host_path(configuration.persistent_domain_root().canonical_label());
let lock = try_exclusive_flock(lock_file(&persistent_domain.join("domain.lock")))?;
let store =
HostCoordinatorDomainStore::open(lock, &persistent_domain.join("domain.redb"), &domain)
.map_err(failure)?;
store
.apply(&[DomainStoreMutation::PutGeneration(Generation {
domain_id: configuration.domain_id(),
generation_id: GenerationId::new(7),
configuration_fingerprint: configuration.configuration_fingerprint(),
coordinator_id: CoordinatorId::new([3; 32]),
network_sidecar_id: NetworkSidecarId::new([4; 32]),
sandbox_id: SandboxId::new([5; 32]),
durable_decision: DurableDecision::None,
})])
.map_err(failure)?;
let coordinator = SandboxCoordinator::new(&store);
coordinator
.begin_intent(
GenerationId::new(7),
container_id,
&runtime_domain.join("runsc"),
BundleDigest::new([7; 32]),
current_process_identity()?,
)
.map_err(failure)?;
Ok(Self {
_tree: tree,
configuration,
store,
bundle,
activity: workspace.join("activity"),
})
}
fn launch_pending(&self) -> Result<AwaitingAttachedRunsc, MatrixError> {
let launch = RuntimeSupervisorLaunch::from_runsc_command(
ProcessOwnershipContext {
domain_id: self.configuration.domain_id(),
generation_id: GenerationId::new(7),
coordinator_id: CoordinatorId::new([3; 32]),
slot: lazaret_core::records::HostProcessOwnershipSlot::SandboxRuntimeSupervisor,
},
self.bundle.command(),
)?;
AwaitingAttachedRunsc::launch(&SandboxCoordinator::new(&self.store), &launch)
.map_err(failure)
}
fn launch_committed(&self) -> Result<CommittedSandboxRuntime, MatrixError> {
let mut pending = self.launch_pending()?;
let staged = pending
.receive_staged(&SandboxCoordinator::new(&self.store))
.map_err(failure)?;
pending
.acknowledge(staged)
.map_err(failure)?
.commit(&SandboxCoordinator::new(&self.store))
.map_err(failure)
}
}
fn stage_rootfs_and_control(runtime_domain: &Path) -> Result<(), MatrixError> {
let rootfs = runtime_domain.join("state-anchor/merged");
for directory in ["bin", "lib64"] {
fs::create_dir_all(rootfs.join(directory))?;
}
for (source, destination) in [
("/bin/sh", "bin/sh"),
("/bin/sleep", "bin/sleep"),
("/lib64/libc.so.6", "lib64/libc.so.6"),
("/lib64/libtinfo.so.6", "lib64/libtinfo.so.6"),
("/lib64/ld-linux-x86-64.so.2", "lib64/ld-linux-x86-64.so.2"),
] {
fs::copy(source, rootfs.join(destination))?;
}
let control = runtime_domain.join("sandbox-control").join(hex(&[6; 32]));
for directory in ["credentials", "mailbox/requests", "mailbox/results", "bin"] {
fs::create_dir_all(control.join(directory))?;
}
fs::copy("/bin/true", control.join("bin/lazaret-escalate"))?;
Ok(())
}
fn sandbox_participant(store: &HostCoordinatorDomainStore) -> lazaret_core::records::Participant {
store
.snapshot()
.unwrap()
.participants
.into_iter()
.find(|participant| participant.role() == ParticipantRole::Sandbox)
.unwrap()
}
fn lock_file(path: &Path) -> File {
OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(path)
.unwrap()
}
fn host_path(path: &HostPath) -> PathBuf {
PathBuf::from(std::ffi::OsString::from_vec(path.as_bytes().to_vec()))
}
fn hex(bytes: &[u8]) -> String {
use std::fmt::Write as _;
bytes.iter().fold(String::new(), |mut output, byte| {
write!(output, "{byte:02x}").unwrap();
output
})
}

12
crates/lazaret-linux/src/ipc.rs
View file

@ -14,6 +14,18 @@ pub fn private_seqpacket_pair() -> std::io::Result<(OwnedFd, OwnedFd)> {
.map_err(Into::into)
}
/// # Errors
/// Returns an error when the private stream pair cannot be created.
pub fn private_stream_pair() -> std::io::Result<(OwnedFd, OwnedFd)> {
socketpair(
AddressFamily::UNIX,
SocketType::STREAM,
SocketFlags::CLOEXEC,
None,
)
.map_err(Into::into)
}
#[cfg(test)]
mod tests {
use std::os::fd::AsFd;

127
crates/lazaret-linux/src/process.rs
View file

@ -1,7 +1,7 @@
use std::ffi::{OsStr, OsString};
use std::fs::File;
use std::io::{self, Read, Write};
use std::os::fd::{AsFd, OwnedFd};
use std::os::fd::{AsFd, BorrowedFd, OwnedFd};
use std::process::Stdio;
use std::sync::{Mutex, OnceLock};
@ -29,6 +29,13 @@ static CHILD_BROKER: OnceLock<ChildBroker> = OnceLock::new();
pub struct TargetCommand {
program: OsString,
arguments: Vec<OsString>,
parent_action: ParentAction,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ParentAction {
Kill,
LifetimePipe,
}
impl TargetCommand {
@ -37,6 +44,7 @@ impl TargetCommand {
Self {
program: program.into(),
arguments: Vec::new(),
parent_action: ParentAction::Kill,
}
}
@ -46,9 +54,46 @@ impl TargetCommand {
self
}
#[must_use]
pub fn with_lifetime_pipe_parent_edge(mut self) -> Self {
self.parent_action = ParentAction::LifetimePipe;
self
}
#[allow(clippy::disallowed_types)]
fn append_to(&self, command: &mut std::process::Command) {
command.arg("--").arg(&self.program).args(&self.arguments);
command
.arg("--parent-action")
.arg(match self.parent_action {
ParentAction::Kill => "kill",
ParentAction::LifetimePipe => "lifetime-pipe",
})
.arg("--")
.arg(&self.program)
.args(&self.arguments);
}
}
pub struct ChildStandardIo {
stdout: Option<Stdio>,
stderr: Option<Stdio>,
}
impl ChildStandardIo {
#[must_use]
pub const fn inherit() -> Self {
Self {
stdout: None,
stderr: None,
}
}
#[must_use]
pub fn redirected(stdout: impl Into<Stdio>, stderr: impl Into<Stdio>) -> Self {
Self {
stdout: Some(stdout.into()),
stderr: Some(stderr.into()),
}
}
}
@ -90,6 +135,12 @@ impl RetainedProcess {
}
}
impl AsFd for RetainedProcess {
fn as_fd(&self) -> BorrowedFd<'_> {
self.pidfd.as_fd()
}
}
#[derive(Debug)]
pub struct ChildBroker {
exclusion: Mutex<()>,
@ -117,6 +168,32 @@ impl ChildBroker {
stage: Stage,
acknowledge: Acknowledge,
) -> io::Result<RetainedProcess>
where
Stage: FnOnce(StagedProcessOwnership) -> io::Result<()>,
Acknowledge: FnOnce(StagedProcessOwnership) -> io::Result<ProcessOwnershipCommitAck>,
{
self.acquire_with_stdio(
expected_parent,
ownership,
target,
ChildStandardIo::inherit(),
stage,
acknowledge,
)
}
/// # Errors
/// Refuses and reaps the blocked wrapper on every failed acquisition or callback.
#[allow(clippy::disallowed_methods, clippy::disallowed_types)]
pub fn acquire_with_stdio<Stage, Acknowledge>(
&self,
expected_parent: PidfsProcessIdentityV1,
ownership: ProcessOwnershipContext,
target: &TargetCommand,
mut standard_io: ChildStandardIo,
stage: Stage,
acknowledge: Acknowledge,
) -> io::Result<RetainedProcess>
where
Stage: FnOnce(StagedProcessOwnership) -> io::Result<()>,
Acknowledge: FnOnce(StagedProcessOwnership) -> io::Result<ProcessOwnershipCommitAck>,
@ -132,6 +209,12 @@ impl ChildBroker {
append_parent_identity(&mut command, expected_parent);
target.append_to(&mut command);
command.stdin(wait.into_stdio());
if let Some(stdout) = standard_io.stdout.take() {
command.stdout(stdout);
}
if let Some(stderr) = standard_io.stderr.take() {
command.stderr(stderr);
}
let mut child = command.spawn()?;
let host_pid = child.id();
let acquired = acquire_spawned_child(host_pid);
@ -217,10 +300,42 @@ fn run_wrapper_role(arguments: Vec<OsString>) -> io::Result<()> {
let (expected_parent, target) = parse_wrapper_arguments(arguments)?;
BarrierWait::from_stdin()?.wait()?;
rustix::io::fcntl_setfd(std::io::stdin(), rustix::io::FdFlags::CLOEXEC)?;
arm_parent_death_and_recheck(Signal::KILL, expected_parent)?;
match target.parent_action {
ParentAction::Kill => {
arm_parent_death_and_recheck(Signal::KILL, expected_parent)?;
}
ParentAction::LifetimePipe => {
require_lifetime_pipe_and_recheck(expected_parent)?;
}
}
Err(exec_target(target))
}
fn require_lifetime_pipe_and_recheck(
expected_parent: PidfsProcessIdentityV1,
) -> io::Result<RetainedProcess> {
let stderr = std::io::stderr();
let mut descriptors = [PollFd::new(&stderr, PollFlags::IN)];
poll(&mut descriptors, Some(&Timespec::default()))?;
if descriptors[0]
.revents()
.intersects(PollFlags::IN | PollFlags::HUP | PollFlags::ERR)
{
return Err(io::Error::other("parent lifetime pipe is not held"));
}
let parent = recover_process(expected_parent)?;
let observed_parent = rustix::process::getppid()
.map(Pid::as_raw_nonzero)
.map(i32::from)
.and_then(|pid| u32::try_from(pid).ok());
if observed_parent != Some(expected_parent.host_pid().get()) {
return Err(io::Error::other(
"parent changed while arming lifetime pipe",
));
}
Ok(parent)
}
#[allow(clippy::disallowed_methods)]
fn acquire_spawned_child(host_pid: u32) -> io::Result<RetainedProcess> {
let raw_pid = i32::try_from(host_pid)
@ -658,6 +773,11 @@ pub fn parse_wrapper_arguments(
let inode = number_argument(&mut arguments, "--parent-inode")?;
let pid = number_argument(&mut arguments, "--parent-pid")?;
let start = number_argument(&mut arguments, "--parent-start")?;
let parent_action = match named_argument(&mut arguments, "--parent-action")?.to_str() {
Some("kill") => ParentAction::Kill,
Some("lifetime-pipe") => ParentAction::LifetimePipe,
_ => return Err(io::Error::other("wrapper parent action is invalid")),
};
if arguments.next().as_deref() != Some(OsStr::new("--")) {
return Err(io::Error::other("wrapper target separator is missing"));
}
@ -667,6 +787,7 @@ pub fn parse_wrapper_arguments(
let target = TargetCommand {
program,
arguments: arguments.collect(),
parent_action,
};
let boot = parse_boot_hex(&boot)?;
Ok((

34
crates/lazaret-linux/tests/acquisition.rs
View file

@ -13,9 +13,10 @@ use lazaret_core::records::{
HostProcessOwnershipSlot, ProcessOwnershipCommitAck, StagedProcessOwnership,
};
use lazaret_linux::process::{
ProcessOwnershipContext, TargetCommand, append_parent_identity, child_broker,
ChildStandardIo, ProcessOwnershipContext, TargetCommand, append_parent_identity, child_broker,
current_process_identity, dispatch_private_process_role, verify_direct_child_pidfd,
};
use rustix::pipe::{PipeFlags, pipe_with};
use rustix::process::{PidfdFlags, pidfd_open};
static SEQUENCE: AtomicU64 = AtomicU64::new(0);
@ -43,6 +44,7 @@ fn exit(result: std::io::Result<()>) -> ExitCode {
fn run_all() -> std::io::Result<()> {
successful_acquisition()?;
lifetime_pipe_acquisition()?;
failed_stages_and_acknowledgments()?;
production_wrapper_refuses_former_harness_path()?;
malformed_barriers_refuse()?;
@ -51,6 +53,32 @@ fn run_all() -> std::io::Result<()> {
concurrent_acquisitions_serialize()
}
#[allow(clippy::disallowed_methods, clippy::disallowed_types)]
fn lifetime_pipe_acquisition() -> std::io::Result<()> {
let marker = marker("lifetime-pipe");
let parent = current_process_identity()?;
let target = TargetCommand::new("/usr/bin/touch")
.argument(&marker)
.with_lifetime_pipe_parent_edge();
let (lifetime_reader, lifetime_writer) = pipe_with(PipeFlags::CLOEXEC)?;
let process = child_broker().acquire_with_stdio(
parent,
context(),
&target,
ChildStandardIo::redirected(Stdio::null(), Stdio::from(lifetime_reader)),
|_| Ok(()),
|staged| Ok(ProcessOwnershipCommitAck::v1(staged)),
)?;
child_broker().reap(process)?;
drop(lifetime_writer);
if !marker.is_file() {
return Err(std::io::Error::other(
"lifetime-pipe acquisition did not execute its target",
));
}
std::fs::remove_file(marker)
}
fn marker(name: &str) -> std::path::PathBuf {
std::env::temp_dir().join(format!(
"lazaret-acquisition-{name}-{}-{}",
@ -243,7 +271,9 @@ fn malformed_barriers_refuse() -> std::io::Result<()> {
.arg("--lazaret-private-process-role")
.arg("launch-wrapper");
append_parent_identity(&mut command, parent);
command.arg("--").arg("/usr/bin/touch").arg(&marker);
command
.args(["--parent-action", "kill", "--", "/usr/bin/touch"])
.arg(&marker);
command.stdin(Stdio::from(temporary_file(&bytes)?));
if command.status()?.success() || marker.exists() {
return Err(std::io::Error::other("malformed barrier released target"));

1
crates/lazaret-runtime-supervisor/Cargo.toml
View file

@ -13,6 +13,7 @@ path = "src/main.rs"
lazaret-core = { version = "=0.0.0", path = "../lazaret-core" }
lazaret-linux = { version = "=0.0.0", path = "../lazaret-linux" }
oci-spec.workspace = true
serde.workspace = true
rustix.workspace = true
serde_json.workspace = true

1
crates/lazaret-runtime-supervisor/src/lib.rs
View file

@ -1,3 +1,4 @@
#![doc = "Runtime construction interfaces owned by the Lazaret Runtime Supervisor."]
pub mod sandbox;
pub mod supervision;

9
crates/lazaret-runtime-supervisor/src/main.rs
View file

@ -7,5 +7,14 @@ fn main() {
}
return;
}
if let Some(result) =
lazaret_runtime_supervisor::supervision::dispatch_private_runtime_role(&arguments)
{
if let Err(error) = result {
eprintln!("Runtime Supervisor failed: {error}");
std::process::exit(1);
}
return;
}
let _ = lazaret_core::PROTOCOL_NAME;
}

2
crates/lazaret-runtime-supervisor/src/sandbox.rs
View file

@ -502,6 +502,8 @@ fn build_runsc_command(paths: &RuntimePaths) -> RunscCommand {
.to_str()
.expect("validated runtime path remains UTF-8")
)),
OsString::from("-rootless"),
OsString::from("-network=host"),
OsString::from("-ignore-cgroups"),
OsString::from("run"),
OsString::from(format!(

625
crates/lazaret-runtime-supervisor/src/supervision.rs Normal file
View file

@ -0,0 +1,625 @@
use std::ffi::{OsStr, OsString};
use std::fs::File;
use std::io;
use std::os::fd::{AsFd, AsRawFd, OwnedFd};
use std::os::unix::fs::FileExt;
use std::process::Stdio;
use std::time::{Duration, Instant};
use lazaret_core::codec::{decode, encode};
use lazaret_core::identity::{CoordinatorId, DomainId, GenerationId};
use lazaret_core::records::{
HostProcessOwnershipSlot, ProcessOwnershipCommitAck, StagedProcessOwnership,
};
use lazaret_linux::capability;
use lazaret_linux::process::{
ChildStandardIo, ProcessOwnershipContext, RetainedProcess, TargetCommand, child_broker,
current_process_identity,
};
use rustix::event::{PollFd, PollFlags, Timespec, poll};
use rustix::fs::{MemfdFlags, memfd_create};
use rustix::io::{FdFlags, fcntl_setfd, read, write};
use rustix::process::Signal;
use rustix::thread::CapabilitySet;
use serde::{Deserialize, Serialize};
use crate::sandbox::RunscCommand;
const PRIVATE_ROLE_ARGUMENT: &str = "--lazaret-runtime-supervisor-role";
const MAX_CONTROL_MESSAGE: usize = 16 * 1024;
const READINESS_TIMEOUT: Duration = Duration::from_secs(30);
const TERMINATION_GRACE: Duration = Duration::from_secs(2);
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum AttachedRunscSignal {
Stop,
Continue,
Kill,
}
impl AttachedRunscSignal {
const fn signal(self) -> Signal {
match self {
Self::Stop => Signal::STOP,
Self::Continue => Signal::CONT,
Self::Kill => Signal::KILL,
}
}
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum SupervisorEvent {
Staged(StagedProcessOwnership),
Ready {
staged: StagedProcessOwnership,
sentry_pid_inventory: u32,
},
SignalApplied(AttachedRunscSignal),
AttachedRunscExited,
Stopped,
Refused(String),
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
enum CoordinatorMessage {
StagedAck(ProcessOwnershipCommitAck),
SandboxCommitted,
SignalAttachedRunsc(AttachedRunscSignal),
Stop,
}
pub struct CoordinatorChannel {
descriptor: OwnedFd,
}
impl CoordinatorChannel {
#[must_use]
pub const fn new(descriptor: OwnedFd) -> Self {
Self { descriptor }
}
/// # Errors
/// Returns an error for a closed, malformed, oversized, or unknown message.
pub fn receive(&self) -> io::Result<SupervisorEvent> {
receive_message(&self.descriptor)
}
/// # Errors
/// Returns an error when the exact typed acknowledgment cannot be delivered.
pub fn acknowledge_staged(&self, acknowledgment: ProcessOwnershipCommitAck) -> io::Result<()> {
send_message(
&self.descriptor,
&CoordinatorMessage::StagedAck(acknowledgment),
)
}
/// # Errors
/// Returns an error when the durable sandbox commit cannot be acknowledged.
pub fn acknowledge_sandbox_committed(&self) -> io::Result<()> {
send_message(&self.descriptor, &CoordinatorMessage::SandboxCommitted)
}
/// # Errors
/// Returns an error when the exact retained-runsc action cannot be requested.
pub fn signal_attached_runsc(&self, signal: AttachedRunscSignal) -> io::Result<()> {
send_message(
&self.descriptor,
&CoordinatorMessage::SignalAttachedRunsc(signal),
)
}
/// # Errors
/// Returns an error when ordered sandbox stop cannot be requested.
pub fn stop(&self) -> io::Result<()> {
send_message(&self.descriptor, &CoordinatorMessage::Stop)
}
}
pub struct RuntimeSupervisorLaunch {
target: TargetCommand,
ownership: ProcessOwnershipContext,
}
impl RuntimeSupervisorLaunch {
/// # Errors
/// Refuses any command other than the deterministic foreground attached-runsc form.
pub fn from_runsc_command(
ownership: ProcessOwnershipContext,
command: &RunscCommand,
) -> io::Result<Self> {
if ownership.slot != HostProcessOwnershipSlot::SandboxRuntimeSupervisor
|| command.program() != OsStr::new("runsc")
|| !command
.arguments()
.iter()
.any(|argument| argument == OsStr::new("run"))
|| command
.arguments()
.iter()
.any(|argument| matches!(argument.to_str(), Some("-detach" | "--detach" | "-d")))
{
return Err(io::Error::other(
"Runtime Supervisor requires foreground attached runsc",
));
}
let mut target = TargetCommand::new("/proc/self/exe")
.argument(PRIVATE_ROLE_ARGUMENT)
.argument("--domain")
.argument(hex(ownership.domain_id.as_bytes()))
.argument("--generation")
.argument(ownership.generation_id.get().to_string())
.argument("--coordinator")
.argument(hex(ownership.coordinator_id.as_bytes()))
.argument("--runsc")
.argument(command.program());
for argument in command.arguments() {
target = target.argument(argument);
}
Ok(Self {
target: target.with_lifetime_pipe_parent_edge(),
ownership,
})
}
#[must_use]
pub const fn target(&self) -> &TargetCommand {
&self.target
}
#[must_use]
pub const fn ownership(&self) -> ProcessOwnershipContext {
self.ownership
}
}
/// Runs the Runtime Supervisor's private early role before ordinary binary dispatch.
///
/// # Errors
/// Returns a fail-closed parse, acquisition, capability, channel, or process error.
#[must_use]
pub fn dispatch_private_runtime_role(arguments: &[OsString]) -> Option<io::Result<()>> {
if arguments
.first()
.is_none_or(|argument| argument != PRIVATE_ROLE_ARGUMENT)
{
return None;
}
let result =
parse_runtime_arguments(&arguments[1..]).and_then(|parsed| run_supervisor(&parsed));
if let Err(error) = &result {
let _ = send_message(
std::io::stdout(),
&SupervisorEvent::Refused(error.to_string()),
);
}
Some(result)
}
struct RuntimeArguments {
ownership: ProcessOwnershipContext,
runsc_program: OsString,
runsc_arguments: Vec<OsString>,
}
fn parse_runtime_arguments(arguments: &[OsString]) -> io::Result<RuntimeArguments> {
let mut arguments = arguments.iter();
let domain = parse_fixed_hex(named(&mut arguments, "--domain")?)?;
let generation = named(&mut arguments, "--generation")?
.to_str()
.and_then(|value| value.parse().ok())
.ok_or_else(|| io::Error::other("Runtime Supervisor generation is invalid"))?;
let coordinator = parse_fixed_hex(named(&mut arguments, "--coordinator")?)?;
if arguments.next().is_none_or(|value| value != "--runsc") {
return Err(io::Error::other(
"Runtime Supervisor runsc separator is missing",
));
}
let program = arguments
.next()
.ok_or_else(|| io::Error::other("Runtime Supervisor runsc program is missing"))?;
if program != "runsc" {
return Err(io::Error::other("Runtime Supervisor target is not runsc"));
}
Ok(RuntimeArguments {
ownership: ProcessOwnershipContext {
domain_id: DomainId::new(domain),
generation_id: GenerationId::new(generation),
coordinator_id: CoordinatorId::new(coordinator),
slot: HostProcessOwnershipSlot::SandboxAttachedRunsc,
},
runsc_program: (*program).clone(),
runsc_arguments: arguments.cloned().collect(),
})
}
fn named<'a>(
arguments: &mut impl Iterator<Item = &'a OsString>,
name: &str,
) -> io::Result<&'a OsString> {
if arguments.next().is_none_or(|value| value != name) {
return Err(io::Error::other(format!(
"Runtime Supervisor argument {name} is missing"
)));
}
arguments
.next()
.ok_or_else(|| io::Error::other(format!("Runtime Supervisor argument {name} has no value")))
}
fn parse_fixed_hex(value: &OsStr) -> io::Result<[u8; 32]> {
let value = value
.to_str()
.ok_or_else(|| io::Error::other("Runtime Supervisor identity is not UTF-8"))?;
if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
return Err(io::Error::other("Runtime Supervisor identity is invalid"));
}
let mut bytes = [0_u8; 32];
for (index, byte) in bytes.iter_mut().enumerate() {
*byte = u8::from_str_radix(&value[index * 2..index * 2 + 2], 16)
.map_err(|_| io::Error::other("Runtime Supervisor identity is invalid"))?;
}
Ok(bytes)
}
fn run_supervisor(arguments: &RuntimeArguments) -> io::Result<()> {
require_lifetime_held()?;
let control = std::io::stdout();
let parent = current_process_identity()?;
let pid_file = File::from(memfd_create("lazaret-runsc-pid", MemfdFlags::empty())?);
fcntl_setfd(pid_file.as_fd(), FdFlags::empty())?;
let runsc = runsc_target(arguments, pid_file.as_raw_fd())?;
let staged = std::cell::Cell::new(None);
let process = child_broker().acquire_with_stdio(
parent,
arguments.ownership,
&runsc,
ChildStandardIo::redirected(Stdio::null(), Stdio::null()),
|ownership| {
staged.set(Some(ownership));
send_message(&control, &SupervisorEvent::Staged(ownership))
},
|ownership| match receive_message(&control)? {
CoordinatorMessage::StagedAck(acknowledgment)
if acknowledgment.acknowledges(&ownership) =>
{
Ok(acknowledgment)
}
_ => Err(io::Error::other(
"Runtime Supervisor received a mismatched staged acknowledgment",
)),
},
)?;
if let Err(error) = fcntl_setfd(pid_file.as_fd(), FdFlags::CLOEXEC) {
return fail_with_exact_cleanup(process, error.into());
}
let Some(staged) = staged.get() else {
return fail_with_exact_cleanup(
process,
io::Error::other("attached runsc ownership was not staged"),
);
};
let sentry_pid_inventory = match wait_for_runsc_ready(&pid_file, &process) {
Ok(pid) => pid,
Err(error) => return fail_with_exact_cleanup(process, error),
};
if let Err(error) = capability::irreversibly_drop(CapabilitySet::NET_ADMIN) {
return fail_with_exact_cleanup(process, error);
}
if let Err(error) = capability::require_regain_refused(CapabilitySet::NET_ADMIN) {
return fail_with_exact_cleanup(process, error);
}
if let Err(error) = send_message(
&control,
&SupervisorEvent::Ready {
staged,
sentry_pid_inventory,
},
) {
return fail_with_exact_cleanup(process, error);
}
let committed: CoordinatorMessage = match receive_message(&control) {
Ok(committed) => committed,
Err(error) => return fail_with_exact_cleanup(process, error),
};
if committed != CoordinatorMessage::SandboxCommitted {
return fail_with_exact_cleanup(
process,
io::Error::other("sandbox commit acknowledgment was not received"),
);
}
supervise(&control, process)
}
fn runsc_target(arguments: &RuntimeArguments, pid_file: i32) -> io::Result<TargetCommand> {
let run = arguments
.runsc_arguments
.iter()
.position(|argument| argument == "run")
.ok_or_else(|| io::Error::other("attached runsc command has no run operation"))?;
let mut target = TargetCommand::new(arguments.runsc_program.clone());
for argument in &arguments.runsc_arguments[..=run] {
target = target.argument(argument);
}
target = target.argument(format!("-pid-file=/proc/self/fd/{pid_file}"));
for argument in &arguments.runsc_arguments[run + 1..] {
target = target.argument(argument);
}
Ok(target)
}
fn wait_for_runsc_ready(pid_file: &File, process: &RetainedProcess) -> io::Result<u32> {
let deadline = Instant::now() + READINESS_TIMEOUT;
let mut bytes = [0_u8; 32];
loop {
require_lifetime_held()?;
if process.has_exited()? {
return Err(io::Error::other("attached runsc exited before readiness"));
}
let length = pid_file.read_at(&mut bytes, 0)?;
if length > 0 {
return std::str::from_utf8(&bytes[..length])
.ok()
.map(str::trim)
.and_then(|value| value.parse().ok())
.filter(|pid| *pid > 0)
.ok_or_else(|| io::Error::other("runsc pid-file readiness is malformed"));
}
if Instant::now() >= deadline {
return Err(io::Error::other("attached runsc readiness timed out"));
}
std::thread::sleep(Duration::from_millis(5));
}
}
fn supervise(control: &std::io::Stdout, process: RetainedProcess) -> io::Result<()> {
loop {
let exited = match process.has_exited() {
Ok(exited) => exited,
Err(error) => return fail_with_exact_cleanup(process, error),
};
if exited {
child_broker().reap(process)?;
send_message(control, &SupervisorEvent::AttachedRunscExited)?;
return Ok(());
}
let lease_lost = match lifetime_lost() {
Ok(lost) => lost,
Err(error) => return fail_with_exact_cleanup(process, error),
};
if lease_lost {
terminate_and_reap(process)?;
return Ok(());
}
let mut descriptors = [PollFd::new(control, PollFlags::IN)];
if let Err(error) = poll(
&mut descriptors,
Some(&Timespec {
tv_sec: 0,
tv_nsec: 10_000_000,
}),
) {
return fail_with_exact_cleanup(process, error.into());
}
let events = descriptors[0].revents();
if events.intersects(PollFlags::HUP | PollFlags::ERR) {
return fail_with_exact_cleanup(
process,
io::Error::new(
io::ErrorKind::UnexpectedEof,
"runtime control channel closed",
),
);
}
if events.intersects(PollFlags::IN) {
let message = match receive_message(control) {
Ok(message) => message,
Err(error) => return fail_with_exact_cleanup(process, error),
};
match message {
CoordinatorMessage::SignalAttachedRunsc(signal) => {
if let Err(error) = process.send_signal(signal.signal()) {
return fail_with_exact_cleanup(process, error);
}
if let Err(error) =
send_message(control, &SupervisorEvent::SignalApplied(signal))
{
return fail_with_exact_cleanup(process, error);
}
}
CoordinatorMessage::Stop => {
terminate_and_reap(process)?;
send_message(control, &SupervisorEvent::Stopped)?;
return Ok(());
}
CoordinatorMessage::StagedAck(_) | CoordinatorMessage::SandboxCommitted => {
return fail_with_exact_cleanup(
process,
io::Error::other("unexpected duplicate coordinator message"),
);
}
}
}
}
}
fn terminate_and_reap(process: RetainedProcess) -> io::Result<()> {
let term_error = process.send_signal(Signal::TERM).err();
let deadline = Instant::now() + TERMINATION_GRACE;
while !process.has_exited()? && Instant::now() < deadline {
std::thread::sleep(Duration::from_millis(5));
}
if !process.has_exited()? {
process.send_signal(Signal::KILL)?;
}
child_broker().reap(process)?;
if let Some(error) = term_error {
Err(error)
} else {
Ok(())
}
}
fn fail_with_exact_cleanup(process: RetainedProcess, error: io::Error) -> io::Result<()> {
let cleanup = terminate_and_reap(process);
match cleanup {
Ok(()) => Err(error),
Err(cleanup) => Err(io::Error::new(
error.kind(),
format!("{error}; exact attached-runsc cleanup failed: {cleanup}"),
)),
}
}
fn require_lifetime_held() -> io::Result<()> {
if lifetime_lost()? {
Err(io::Error::other("Host Coordinator lifetime lease is lost"))
} else {
Ok(())
}
}
fn lifetime_lost() -> io::Result<bool> {
let stderr = std::io::stderr();
let mut descriptors = [PollFd::new(&stderr, PollFlags::IN)];
poll(&mut descriptors, Some(&Timespec::default()))?;
Ok(descriptors[0]
.revents()
.intersects(PollFlags::IN | PollFlags::HUP | PollFlags::ERR))
}
fn send_message<T: Serialize>(descriptor: impl AsFd, message: &T) -> io::Result<()> {
let encoded = encode(message).map_err(|error| io::Error::other(error.to_string()))?;
if encoded.len() > MAX_CONTROL_MESSAGE {
return Err(io::Error::other("runtime control message is oversized"));
}
let length = u32::try_from(encoded.len())
.map_err(|_| io::Error::other("runtime control message is oversized"))?
.to_be_bytes();
send_all(descriptor.as_fd(), &length)?;
send_all(descriptor.as_fd(), &encoded)
}
fn receive_message<T>(descriptor: impl AsFd) -> io::Result<T>
where
T: serde::de::DeserializeOwned + Serialize,
{
let mut length = [0_u8; 4];
receive_exact(descriptor.as_fd(), &mut length)?;
let length = usize::try_from(u32::from_be_bytes(length))
.map_err(|_| io::Error::other("runtime control length is invalid"))?;
if length > MAX_CONTROL_MESSAGE {
return Err(io::Error::other("runtime control message is oversized"));
}
let mut encoded = vec![0_u8; length];
receive_exact(descriptor.as_fd(), &mut encoded)?;
decode(&encoded).map_err(|error| io::Error::other(error.to_string()))
}
fn send_all(descriptor: impl AsFd + Copy, mut bytes: &[u8]) -> io::Result<()> {
while !bytes.is_empty() {
let sent = write(descriptor, bytes)?;
if sent == 0 {
return Err(io::Error::new(
io::ErrorKind::WriteZero,
"runtime control channel closed",
));
}
bytes = &bytes[sent..];
}
Ok(())
}
fn receive_exact(descriptor: impl AsFd + Copy, mut bytes: &mut [u8]) -> io::Result<()> {
while !bytes.is_empty() {
let received = read(descriptor, &mut *bytes)?;
if received == 0 {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"runtime control channel closed",
));
}
bytes = &mut bytes[received..];
}
Ok(())
}
fn hex(bytes: &[u8]) -> String {
use std::fmt::Write as _;
bytes.iter().fold(String::new(), |mut output, byte| {
write!(output, "{byte:02x}").expect("writing to a String cannot fail");
output
})
}
#[cfg(test)]
mod tests {
use std::ffi::OsString;
use lazaret_core::identity::{CoordinatorId, DomainId, GenerationId};
use lazaret_core::records::HostProcessOwnershipSlot;
use lazaret_linux::ipc::private_stream_pair;
use super::*;
#[test]
fn private_stream_framing_round_trips_typed_events() {
let (coordinator, supervisor) = private_stream_pair().unwrap();
let expected = SupervisorEvent::Refused("refused".into());
send_message(&supervisor, &expected).unwrap();
assert_eq!(
CoordinatorChannel::new(coordinator).receive().unwrap(),
expected
);
}
#[test]
fn runtime_role_parser_preserves_exact_ownership_and_runsc_arguments() {
let domain = [1_u8; 32];
let coordinator = [2_u8; 32];
let arguments = [
"--domain".into(),
hex(&domain).into(),
"--generation".into(),
"7".into(),
"--coordinator".into(),
hex(&coordinator).into(),
"--runsc".into(),
"runsc".into(),
"-root=/runtime".into(),
"run".into(),
"container".into(),
];
let parsed = parse_runtime_arguments(&arguments).unwrap();
assert_eq!(parsed.ownership.domain_id, DomainId::new(domain));
assert_eq!(parsed.ownership.generation_id, GenerationId::new(7));
assert_eq!(
parsed.ownership.coordinator_id,
CoordinatorId::new(coordinator)
);
assert_eq!(
parsed.ownership.slot,
HostProcessOwnershipSlot::SandboxAttachedRunsc
);
assert_eq!(parsed.runsc_program, OsString::from("runsc"));
assert_eq!(
parsed.runsc_arguments,
["-root=/runtime", "run", "container"].map(OsString::from)
);
}
#[test]
fn runtime_role_parser_refuses_non_runsc_target() {
let arguments = [
"--domain".into(),
hex(&[1_u8; 32]).into(),
"--generation".into(),
"7".into(),
"--coordinator".into(),
hex(&[2_u8; 32]).into(),
"--runsc".into(),
"foreign-runtime".into(),
];
assert!(parse_runtime_arguments(&arguments).is_err());
}
}

2
crates/lazaret-runtime-supervisor/tests/sandbox_builder.rs
View file

@ -234,6 +234,8 @@ fn deterministic_snapshot_has_only_private_namespaces_and_fixed_runtime_shape()
.collect::<Vec<_>>(),
[
format!("-root={runtime_domain}/runsc"),
"-rootless".to_owned(),
"-network=host".to_owned(),
"-ignore-cgroups".to_owned(),
"run".to_owned(),
format!(