786 lines
28 KiB
Rust
786 lines
28 KiB
Rust
use std::fmt;
|
|
use std::fs::File;
|
|
use std::io::{self, Read, Write};
|
|
use std::os::fd::AsFd;
|
|
use std::os::unix::ffi::OsStringExt;
|
|
use std::os::unix::fs::{FileTypeExt, MetadataExt, PermissionsExt};
|
|
use std::os::unix::net::{UnixListener, UnixStream};
|
|
use std::path::PathBuf;
|
|
|
|
use lazaret_core::codec;
|
|
use lazaret_core::configuration::ResolvedDomainConfiguration;
|
|
use lazaret_core::identity::{
|
|
ClientId, ConfigurationFingerprint, DomainId, GenerationId, SessionAnchorId, SessionId,
|
|
};
|
|
use lazaret_core::records::{ProtocolVersion, Session, SessionPhase};
|
|
use rustix::net::sockopt::socket_peercred;
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
use crate::lifecycle::{HostCoordinatorLifecycle, LifecycleAdapter, SessionLifecycleAdapter};
|
|
|
|
pub const BOOTSTRAP_WAITER_SOCKET_NAME: &str = "wait.sock";
|
|
const MAX_FRAME_BYTES: usize = 4_096;
|
|
const MAX_WAITERS: usize = 64;
|
|
|
|
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
|
pub struct BootstrapWaitRequest {
|
|
pub protocol_version: ProtocolVersion,
|
|
pub domain_id: DomainId,
|
|
pub configuration_fingerprint: ConfigurationFingerprint,
|
|
pub cli_id: ClientId,
|
|
pub session_id: SessionId,
|
|
pub session_anchor_id: SessionAnchorId,
|
|
}
|
|
|
|
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
|
pub enum BootstrapWaitOutcome {
|
|
Waiting,
|
|
Published,
|
|
Failed,
|
|
Refused,
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct BootstrapWaiterError(String);
|
|
|
|
impl fmt::Display for BootstrapWaiterError {
|
|
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
|
|
formatter.write_str(&self.0)
|
|
}
|
|
}
|
|
|
|
impl std::error::Error for BootstrapWaiterError {}
|
|
|
|
fn failure(error: impl fmt::Display) -> BootstrapWaiterError {
|
|
BootstrapWaiterError(error.to_string())
|
|
}
|
|
|
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
|
struct FileIdentity {
|
|
device: u64,
|
|
inode: u64,
|
|
}
|
|
|
|
impl FileIdentity {
|
|
fn from_metadata(metadata: &std::fs::Metadata) -> Self {
|
|
Self {
|
|
device: metadata.dev(),
|
|
inode: metadata.ino(),
|
|
}
|
|
}
|
|
}
|
|
|
|
struct PendingWaiter {
|
|
stream: UnixStream,
|
|
bytes: Vec<u8>,
|
|
}
|
|
|
|
struct LiveWaiter {
|
|
stream: UnixStream,
|
|
session_id: SessionId,
|
|
}
|
|
|
|
pub struct BootstrapWaiterFace {
|
|
root: File,
|
|
root_path: PathBuf,
|
|
root_identity: FileIdentity,
|
|
socket_path: PathBuf,
|
|
socket_identity: FileIdentity,
|
|
listener: UnixListener,
|
|
pending: Vec<PendingWaiter>,
|
|
waiters: Vec<LiveWaiter>,
|
|
domain_id: DomainId,
|
|
configuration_fingerprint: ConfigurationFingerprint,
|
|
generation_id: GenerationId,
|
|
uid: u32,
|
|
withdrawn: bool,
|
|
}
|
|
|
|
impl BootstrapWaiterFace {
|
|
/// Publishes one bounded invoking-user-only waiter admission socket.
|
|
///
|
|
/// # Errors
|
|
/// Refuses changed Runtime Domain authority or a preexisting, replaced, or non-private path.
|
|
pub fn bind(
|
|
configuration: &ResolvedDomainConfiguration,
|
|
generation_id: GenerationId,
|
|
uid: u32,
|
|
) -> Result<Self, BootstrapWaiterError> {
|
|
let root = File::from(
|
|
configuration
|
|
.runtime_domain_root()
|
|
.open_authority()
|
|
.map_err(failure)?,
|
|
);
|
|
let root_path = PathBuf::from(std::ffi::OsString::from_vec(
|
|
configuration
|
|
.runtime_domain_root()
|
|
.canonical_label()
|
|
.as_bytes()
|
|
.to_vec(),
|
|
));
|
|
let root_metadata = root.metadata().map_err(failure)?;
|
|
let path_metadata = std::fs::symlink_metadata(&root_path).map_err(failure)?;
|
|
if FileIdentity::from_metadata(&root_metadata)
|
|
!= FileIdentity::from_metadata(&path_metadata)
|
|
|| !root_metadata.is_dir()
|
|
|| root_metadata.uid() != uid
|
|
|| root_metadata.mode() & 0o7777 != 0o700
|
|
{
|
|
return Err(failure(
|
|
"Runtime Domain root is not exact invoking-user mode 0700",
|
|
));
|
|
}
|
|
let socket_path = root_path.join(BOOTSTRAP_WAITER_SOCKET_NAME);
|
|
match std::fs::symlink_metadata(&socket_path) {
|
|
Err(error) if error.kind() == io::ErrorKind::NotFound => {}
|
|
Ok(_) => return Err(failure("bootstrap waiter path already exists")),
|
|
Err(error) => return Err(failure(error)),
|
|
}
|
|
let listener = UnixListener::bind(&socket_path).map_err(failure)?;
|
|
std::fs::set_permissions(&socket_path, std::fs::Permissions::from_mode(0o600))
|
|
.map_err(failure)?;
|
|
listener.set_nonblocking(true).map_err(failure)?;
|
|
let socket_metadata = std::fs::symlink_metadata(&socket_path).map_err(failure)?;
|
|
if !socket_metadata.file_type().is_socket()
|
|
|| socket_metadata.uid() != uid
|
|
|| socket_metadata.mode() & 0o7777 != 0o600
|
|
{
|
|
return Err(failure(
|
|
"bootstrap waiter socket is not invoking-user mode 0600",
|
|
));
|
|
}
|
|
Ok(Self {
|
|
root,
|
|
root_path,
|
|
root_identity: FileIdentity::from_metadata(&root_metadata),
|
|
socket_path,
|
|
socket_identity: FileIdentity::from_metadata(&socket_metadata),
|
|
listener,
|
|
pending: Vec::new(),
|
|
waiters: Vec::new(),
|
|
domain_id: configuration.domain_id(),
|
|
configuration_fingerprint: configuration.configuration_fingerprint(),
|
|
generation_id,
|
|
uid,
|
|
withdrawn: false,
|
|
})
|
|
}
|
|
|
|
/// Admits every complete valid request currently available without blocking.
|
|
///
|
|
/// # Errors
|
|
/// Refuses changed socket authority, I/O failures, or durable lifecycle failures.
|
|
pub fn admit_pending<A, S>(
|
|
&mut self,
|
|
lifecycle: &mut HostCoordinatorLifecycle<'_, A, S>,
|
|
) -> Result<(), BootstrapWaiterError>
|
|
where
|
|
A: LifecycleAdapter,
|
|
A::Error: fmt::Display,
|
|
S: SessionLifecycleAdapter<Error = A::Error>,
|
|
{
|
|
self.validate_paths()?;
|
|
while self.pending.len() + self.waiters.len() < MAX_WAITERS {
|
|
match self.listener.accept() {
|
|
Ok((stream, _)) => {
|
|
let credentials = socket_peercred(stream.as_fd()).map_err(failure)?;
|
|
if credentials.uid.as_raw() != self.uid {
|
|
continue;
|
|
}
|
|
stream.set_nonblocking(true).map_err(failure)?;
|
|
self.pending.push(PendingWaiter {
|
|
stream,
|
|
bytes: Vec::new(),
|
|
});
|
|
}
|
|
Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
|
|
Err(error) => return Err(failure(error)),
|
|
}
|
|
}
|
|
|
|
let mut index = 0;
|
|
while index < self.pending.len() {
|
|
match read_request(&mut self.pending[index]) {
|
|
Ok(Some(request)) => {
|
|
let mut pending = self.pending.swap_remove(index);
|
|
if request.protocol_version != ProtocolVersion::V1
|
|
|| request.domain_id != self.domain_id
|
|
|| request.configuration_fingerprint != self.configuration_fingerprint
|
|
{
|
|
let _ = send_outcome(&mut pending.stream, BootstrapWaitOutcome::Refused);
|
|
continue;
|
|
}
|
|
let session = Session {
|
|
generation_id: self.generation_id,
|
|
session_id: request.session_id,
|
|
cli_id: request.cli_id,
|
|
session_anchor_id: request.session_anchor_id,
|
|
session_phase: SessionPhase::New,
|
|
};
|
|
if lifecycle.wait_for_domain(session).is_err() {
|
|
let _ = send_outcome(&mut pending.stream, BootstrapWaitOutcome::Refused);
|
|
continue;
|
|
}
|
|
if send_outcome(&mut pending.stream, BootstrapWaitOutcome::Waiting).is_err() {
|
|
lifecycle
|
|
.disconnect_session(request.session_id)
|
|
.map_err(failure)?;
|
|
continue;
|
|
}
|
|
self.waiters.push(LiveWaiter {
|
|
stream: pending.stream,
|
|
session_id: request.session_id,
|
|
});
|
|
}
|
|
Ok(None) => index += 1,
|
|
Err(_) => {
|
|
self.pending.swap_remove(index);
|
|
}
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Withdraws the exact waiter admission path.
|
|
///
|
|
/// # Errors
|
|
/// Refuses changed authority or failure to remove the exact socket.
|
|
pub fn withdraw(&mut self) -> Result<(), BootstrapWaiterError> {
|
|
if self.withdrawn {
|
|
return Ok(());
|
|
}
|
|
self.validate_paths()?;
|
|
std::fs::remove_file(&self.socket_path).map_err(failure)?;
|
|
self.withdrawn = true;
|
|
Ok(())
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn notify(&mut self, outcome: BootstrapWaitOutcome) -> Vec<SessionId> {
|
|
if matches!(
|
|
outcome,
|
|
BootstrapWaitOutcome::Waiting | BootstrapWaitOutcome::Refused
|
|
) {
|
|
return self
|
|
.waiters
|
|
.iter()
|
|
.map(|waiter| waiter.session_id)
|
|
.collect();
|
|
}
|
|
let mut failed = Vec::new();
|
|
self.waiters.retain_mut(|waiter| {
|
|
let retained = send_outcome(&mut waiter.stream, outcome).is_ok()
|
|
&& outcome == BootstrapWaitOutcome::Published
|
|
&& waiter.stream.set_nonblocking(true).is_ok();
|
|
if !retained && outcome == BootstrapWaitOutcome::Published {
|
|
failed.push(waiter.session_id);
|
|
}
|
|
retained
|
|
});
|
|
failed
|
|
}
|
|
|
|
/// Reconciles published waiter streams with their durable control-registration phase.
|
|
///
|
|
/// # Errors
|
|
/// Returns a durable lifecycle error when a vanished Waiting session cannot be disconnected.
|
|
pub fn reconcile_published<A, S>(
|
|
&mut self,
|
|
lifecycle: &mut HostCoordinatorLifecycle<'_, A, S>,
|
|
) -> Result<(), BootstrapWaiterError>
|
|
where
|
|
A: LifecycleAdapter,
|
|
A::Error: fmt::Display,
|
|
S: SessionLifecycleAdapter<Error = A::Error>,
|
|
{
|
|
let mut index = 0;
|
|
while index < self.waiters.len() {
|
|
let closed = !matches!(
|
|
self.waiters[index].stream.read(&mut [0_u8; 1]),
|
|
Err(error) if error.kind() == io::ErrorKind::WouldBlock
|
|
);
|
|
if !closed {
|
|
index += 1;
|
|
continue;
|
|
}
|
|
let waiter = self.waiters.swap_remove(index);
|
|
if lifecycle.session_phase(waiter.session_id) == Some(SessionPhase::Waiting) {
|
|
lifecycle
|
|
.disconnect_session(waiter.session_id)
|
|
.map_err(failure)?;
|
|
lifecycle.teardown_if_idle().map_err(failure)?;
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn validate_paths(&self) -> Result<(), BootstrapWaiterError> {
|
|
let root_metadata = self.root.metadata().map_err(failure)?;
|
|
let path_metadata = std::fs::symlink_metadata(&self.root_path).map_err(failure)?;
|
|
let socket_metadata = std::fs::symlink_metadata(&self.socket_path).map_err(failure)?;
|
|
if FileIdentity::from_metadata(&root_metadata) != self.root_identity
|
|
|| FileIdentity::from_metadata(&path_metadata) != self.root_identity
|
|
|| !root_metadata.is_dir()
|
|
|| root_metadata.uid() != self.uid
|
|
|| root_metadata.mode() & 0o7777 != 0o700
|
|
|| FileIdentity::from_metadata(&socket_metadata) != self.socket_identity
|
|
|| !socket_metadata.file_type().is_socket()
|
|
|| socket_metadata.uid() != self.uid
|
|
|| socket_metadata.mode() & 0o7777 != 0o600
|
|
{
|
|
return Err(failure("bootstrap waiter authority changed"));
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
impl Drop for BootstrapWaiterFace {
|
|
fn drop(&mut self) {
|
|
if !self.withdrawn
|
|
&& std::fs::symlink_metadata(&self.socket_path)
|
|
.ok()
|
|
.is_some_and(|metadata| {
|
|
FileIdentity::from_metadata(&metadata) == self.socket_identity
|
|
})
|
|
{
|
|
let _ = std::fs::remove_file(&self.socket_path);
|
|
}
|
|
}
|
|
}
|
|
|
|
fn read_request(
|
|
pending: &mut PendingWaiter,
|
|
) -> Result<Option<BootstrapWaitRequest>, BootstrapWaiterError> {
|
|
let mut buffer = [0_u8; 1_024];
|
|
loop {
|
|
match pending.stream.read(&mut buffer) {
|
|
Ok(0) => return Err(failure("bootstrap waiter closed before one request")),
|
|
Ok(length) => {
|
|
pending.bytes.extend_from_slice(&buffer[..length]);
|
|
if pending.bytes.len() > MAX_FRAME_BYTES + 4 {
|
|
return Err(failure("bootstrap waiter request is oversized"));
|
|
}
|
|
}
|
|
Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
|
|
Err(error) => return Err(failure(error)),
|
|
}
|
|
}
|
|
if pending.bytes.len() < 4 {
|
|
return Ok(None);
|
|
}
|
|
let length = u32::from_be_bytes(pending.bytes[..4].try_into().expect("four bytes")) as usize;
|
|
if length > MAX_FRAME_BYTES {
|
|
return Err(failure("bootstrap waiter request is oversized"));
|
|
}
|
|
if pending.bytes.len() < length + 4 {
|
|
return Ok(None);
|
|
}
|
|
if pending.bytes.len() != length + 4 {
|
|
return Err(failure("bootstrap waiter request has trailing bytes"));
|
|
}
|
|
codec::decode::<BootstrapWaitRequest>(&pending.bytes[4..])
|
|
.map(Some)
|
|
.map_err(failure)
|
|
}
|
|
|
|
fn send_outcome(
|
|
stream: &mut UnixStream,
|
|
outcome: BootstrapWaitOutcome,
|
|
) -> Result<(), BootstrapWaiterError> {
|
|
stream.set_nonblocking(false).map_err(failure)?;
|
|
write_frame(stream, &outcome)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses oversized or unencodable records and stream write failures.
|
|
pub fn write_frame<T: Serialize>(
|
|
stream: &mut UnixStream,
|
|
record: &T,
|
|
) -> Result<(), BootstrapWaiterError> {
|
|
let encoded = codec::encode(record).map_err(failure)?;
|
|
if encoded.len() > MAX_FRAME_BYTES {
|
|
return Err(failure("bootstrap waiter frame is oversized"));
|
|
}
|
|
let length = u32::try_from(encoded.len()).map_err(failure)?;
|
|
stream.write_all(&length.to_be_bytes()).map_err(failure)?;
|
|
stream.write_all(&encoded).map_err(failure)?;
|
|
stream.flush().map_err(failure)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses oversized, malformed, or incomplete frames and stream read failures.
|
|
pub fn read_frame<T>(stream: &mut UnixStream) -> Result<T, BootstrapWaiterError>
|
|
where
|
|
T: serde::de::DeserializeOwned + Serialize,
|
|
{
|
|
let mut length = [0_u8; 4];
|
|
stream.read_exact(&mut length).map_err(failure)?;
|
|
let length = u32::from_be_bytes(length) as usize;
|
|
if length > MAX_FRAME_BYTES {
|
|
return Err(failure("bootstrap waiter frame is oversized"));
|
|
}
|
|
let mut encoded = vec![0_u8; length];
|
|
stream.read_exact(&mut encoded).map_err(failure)?;
|
|
codec::decode(&encoded).map_err(failure)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use std::collections::BTreeMap;
|
|
use std::fs::{self, OpenOptions};
|
|
use std::sync::Arc;
|
|
use std::thread;
|
|
|
|
use lazaret_core::configuration::{ConfigurationLoadRequest, load_trusted_configuration};
|
|
use lazaret_core::identity::{CoordinatorId, NetworkSidecarId, SandboxId};
|
|
use lazaret_core::records::{Domain, DurableDecision, Generation};
|
|
use lazaret_linux::filesystem::try_exclusive_flock;
|
|
use lazaret_runtime_supervisor::sandbox::default_sandbox_layout;
|
|
|
|
use crate::domain_store::{DomainStoreMutation, HostCoordinatorDomainStore};
|
|
use crate::lifecycle::{LifecycleAction, SessionAdmissionIdentity};
|
|
|
|
use super::*;
|
|
|
|
struct Adapter;
|
|
|
|
impl LifecycleAdapter for Adapter {
|
|
type Error = &'static str;
|
|
|
|
fn execute(
|
|
&mut self,
|
|
_action: LifecycleAction,
|
|
_generation: &Generation,
|
|
) -> Result<(), Self::Error> {
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
struct Admission(Session, ConfigurationFingerprint);
|
|
|
|
struct Sessions;
|
|
|
|
impl SessionLifecycleAdapter for Sessions {
|
|
type Error = &'static str;
|
|
type Admission = Admission;
|
|
|
|
fn admission_identity(&self, admission: &Admission) -> SessionAdmissionIdentity {
|
|
SessionAdmissionIdentity {
|
|
generation_id: admission.0.generation_id,
|
|
session_id: admission.0.session_id,
|
|
configuration_fingerprint: admission.1,
|
|
}
|
|
}
|
|
|
|
fn register_session(
|
|
&self,
|
|
store: &HostCoordinatorDomainStore,
|
|
_entry_action: LifecycleAction,
|
|
admission: Admission,
|
|
) -> Result<Session, Self::Error> {
|
|
let registering = Session {
|
|
session_phase: SessionPhase::Registering,
|
|
..admission.0
|
|
};
|
|
let attached = Session {
|
|
session_phase: SessionPhase::Attached,
|
|
..registering.clone()
|
|
};
|
|
store
|
|
.apply(&[
|
|
DomainStoreMutation::PutSession(registering),
|
|
DomainStoreMutation::PutSession(attached.clone()),
|
|
])
|
|
.map_err(|_| "register")?;
|
|
Ok(attached)
|
|
}
|
|
|
|
fn disconnect_session(
|
|
&self,
|
|
store: &HostCoordinatorDomainStore,
|
|
session: &Session,
|
|
) -> Result<(), Self::Error> {
|
|
store
|
|
.apply(&[DomainStoreMutation::PutSession(Session {
|
|
session_phase: SessionPhase::Closed,
|
|
..session.clone()
|
|
})])
|
|
.map_err(|_| "disconnect")
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn waiter_is_durable_before_notification_and_advances_after_publish() {
|
|
let tree = tempfile::Builder::new()
|
|
.prefix("lz-")
|
|
.tempdir_in("/tmp")
|
|
.unwrap();
|
|
let configuration = configuration(tree.path());
|
|
let store = store(&configuration);
|
|
let mut lifecycle = HostCoordinatorLifecycle::new(&store, Adapter, Sessions);
|
|
lifecycle
|
|
.start_new_domain(generation(&configuration), session(1))
|
|
.unwrap();
|
|
let mut face = BootstrapWaiterFace::bind(
|
|
&configuration,
|
|
GenerationId::new(1),
|
|
rustix::process::getuid().as_raw(),
|
|
)
|
|
.unwrap();
|
|
let socket = face.socket_path.clone();
|
|
let request = BootstrapWaitRequest {
|
|
protocol_version: ProtocolVersion::V1,
|
|
domain_id: configuration.domain_id(),
|
|
configuration_fingerprint: configuration.configuration_fingerprint(),
|
|
cli_id: ClientId::new([34; 32]),
|
|
session_id: SessionId::new([2; 32]),
|
|
session_anchor_id: SessionAnchorId::new([66; 32]),
|
|
};
|
|
let client_request = request.clone();
|
|
let client = thread::spawn(move || {
|
|
let mut stream = UnixStream::connect(socket).unwrap();
|
|
write_frame(&mut stream, &client_request).unwrap();
|
|
assert_eq!(
|
|
read_frame::<BootstrapWaitOutcome>(&mut stream).unwrap(),
|
|
BootstrapWaitOutcome::Waiting
|
|
);
|
|
assert_eq!(
|
|
read_frame::<BootstrapWaitOutcome>(&mut stream).unwrap(),
|
|
BootstrapWaitOutcome::Published
|
|
);
|
|
});
|
|
for _ in 0..10_000 {
|
|
face.admit_pending(&mut lifecycle).unwrap();
|
|
if lifecycle.session_phase(request.session_id).is_some() {
|
|
break;
|
|
}
|
|
thread::yield_now();
|
|
}
|
|
assert!(lifecycle.session_phase(request.session_id).is_some());
|
|
assert_eq!(
|
|
store
|
|
.snapshot()
|
|
.unwrap()
|
|
.sessions
|
|
.iter()
|
|
.find(|session| session.session_id == request.session_id)
|
|
.unwrap()
|
|
.session_phase,
|
|
SessionPhase::Waiting
|
|
);
|
|
complete_startup(&mut lifecycle);
|
|
face.withdraw().unwrap();
|
|
assert!(face.notify(BootstrapWaitOutcome::Published).is_empty());
|
|
lifecycle
|
|
.admit_session(Admission(
|
|
Session {
|
|
generation_id: GenerationId::new(1),
|
|
session_id: request.session_id,
|
|
cli_id: request.cli_id,
|
|
session_anchor_id: request.session_anchor_id,
|
|
session_phase: SessionPhase::Waiting,
|
|
},
|
|
configuration.configuration_fingerprint(),
|
|
))
|
|
.unwrap();
|
|
assert_eq!(
|
|
lifecycle.session_phase(request.session_id),
|
|
Some(SessionPhase::Attached)
|
|
);
|
|
client.join().unwrap();
|
|
face.reconcile_published(&mut lifecycle).unwrap();
|
|
assert_eq!(
|
|
lifecycle.session_phase(request.session_id),
|
|
Some(SessionPhase::Attached)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn vanished_published_waiter_releases_its_durable_interest() {
|
|
let tree = tempfile::Builder::new()
|
|
.prefix("lz-")
|
|
.tempdir_in("/tmp")
|
|
.unwrap();
|
|
let configuration = configuration(tree.path());
|
|
let store = store(&configuration);
|
|
let mut lifecycle = HostCoordinatorLifecycle::new(&store, Adapter, Sessions);
|
|
lifecycle
|
|
.start_new_domain(generation(&configuration), session(1))
|
|
.unwrap();
|
|
let mut face = BootstrapWaiterFace::bind(
|
|
&configuration,
|
|
GenerationId::new(1),
|
|
rustix::process::getuid().as_raw(),
|
|
)
|
|
.unwrap();
|
|
let request = BootstrapWaitRequest {
|
|
protocol_version: ProtocolVersion::V1,
|
|
domain_id: configuration.domain_id(),
|
|
configuration_fingerprint: configuration.configuration_fingerprint(),
|
|
cli_id: ClientId::new([35; 32]),
|
|
session_id: SessionId::new([3; 32]),
|
|
session_anchor_id: SessionAnchorId::new([67; 32]),
|
|
};
|
|
let socket = face.socket_path.clone();
|
|
let client_request = request.clone();
|
|
let client = thread::spawn(move || {
|
|
let mut stream = UnixStream::connect(socket).unwrap();
|
|
write_frame(&mut stream, &client_request).unwrap();
|
|
assert_eq!(
|
|
read_frame::<BootstrapWaitOutcome>(&mut stream).unwrap(),
|
|
BootstrapWaitOutcome::Waiting
|
|
);
|
|
assert_eq!(
|
|
read_frame::<BootstrapWaitOutcome>(&mut stream).unwrap(),
|
|
BootstrapWaitOutcome::Published
|
|
);
|
|
});
|
|
for _ in 0..10_000 {
|
|
face.admit_pending(&mut lifecycle).unwrap();
|
|
if lifecycle.session_phase(request.session_id) == Some(SessionPhase::Waiting) {
|
|
break;
|
|
}
|
|
thread::yield_now();
|
|
}
|
|
complete_startup(&mut lifecycle);
|
|
face.withdraw().unwrap();
|
|
assert!(face.notify(BootstrapWaitOutcome::Published).is_empty());
|
|
client.join().unwrap();
|
|
for _ in 0..10_000 {
|
|
face.reconcile_published(&mut lifecycle).unwrap();
|
|
if lifecycle.session_phase(request.session_id) == Some(SessionPhase::Closed) {
|
|
break;
|
|
}
|
|
thread::yield_now();
|
|
}
|
|
assert_eq!(
|
|
lifecycle.session_phase(request.session_id),
|
|
Some(SessionPhase::Closed)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn replacement_cannot_receive_or_be_removed_as_waiter_authority() {
|
|
let tree = tempfile::Builder::new()
|
|
.prefix("lz-")
|
|
.tempdir_in("/tmp")
|
|
.unwrap();
|
|
let configuration = configuration(tree.path());
|
|
let store = store(&configuration);
|
|
let mut lifecycle = HostCoordinatorLifecycle::new(&store, Adapter, Sessions);
|
|
lifecycle
|
|
.start_new_domain(generation(&configuration), session(1))
|
|
.unwrap();
|
|
let mut face = BootstrapWaiterFace::bind(
|
|
&configuration,
|
|
GenerationId::new(1),
|
|
rustix::process::getuid().as_raw(),
|
|
)
|
|
.unwrap();
|
|
let replacement = face.socket_path.with_extension("old");
|
|
fs::rename(&face.socket_path, replacement).unwrap();
|
|
let replacement_listener = UnixListener::bind(&face.socket_path).unwrap();
|
|
fs::set_permissions(&face.socket_path, fs::Permissions::from_mode(0o600)).unwrap();
|
|
assert!(face.admit_pending(&mut lifecycle).is_err());
|
|
let path = face.socket_path.clone();
|
|
drop(face);
|
|
assert!(path.exists());
|
|
drop(replacement_listener);
|
|
}
|
|
|
|
fn complete_startup(lifecycle: &mut HostCoordinatorLifecycle<'_, Adapter, Sessions>) {
|
|
lifecycle.create_trust_anchor().unwrap();
|
|
lifecycle.start_runtime_supervisor().unwrap();
|
|
lifecycle.lease_prepared_lower().unwrap();
|
|
lifecycle.mount_state_anchor().unwrap();
|
|
lifecycle.start_sidecar().unwrap();
|
|
lifecycle.start_connectivity_helper().unwrap();
|
|
lifecycle.mark_network_ready().unwrap();
|
|
lifecycle.arm_runtime_causal_edge().unwrap();
|
|
lifecycle.start_sandbox().unwrap();
|
|
lifecycle.publish_domain().unwrap();
|
|
}
|
|
|
|
fn store(configuration: &ResolvedDomainConfiguration) -> Arc<HostCoordinatorDomainStore> {
|
|
let root = PathBuf::from(std::ffi::OsString::from_vec(
|
|
configuration
|
|
.persistent_domain_root()
|
|
.canonical_label()
|
|
.as_bytes()
|
|
.to_vec(),
|
|
));
|
|
let lock = OpenOptions::new()
|
|
.read(true)
|
|
.write(true)
|
|
.create(true)
|
|
.truncate(false)
|
|
.open(root.join("domain.lock"))
|
|
.unwrap();
|
|
Arc::new(
|
|
HostCoordinatorDomainStore::open(
|
|
try_exclusive_flock(lock).unwrap(),
|
|
&root.join("domain.redb"),
|
|
&Domain {
|
|
domain_id: configuration.domain_id(),
|
|
profile_name: "agent".into(),
|
|
canonical_workspace: configuration.workspace().canonical_label().clone(),
|
|
},
|
|
)
|
|
.unwrap(),
|
|
)
|
|
}
|
|
|
|
fn configuration(root: &std::path::Path) -> ResolvedDomainConfiguration {
|
|
let configuration_directory = root.join("configuration");
|
|
let workspace = root.join("workspace");
|
|
let runtime = root.join("r");
|
|
fs::create_dir(&configuration_directory).unwrap();
|
|
fs::create_dir(&workspace).unwrap();
|
|
fs::create_dir(&runtime).unwrap();
|
|
let configuration_path = configuration_directory.join("lazaret.toml");
|
|
fs::write(
|
|
&configuration_path,
|
|
format!(
|
|
"schema = \"lazaret-trusted-host-config-v1\"\n[storage]\ndata_root = \"{}\"\n[network]\ndefault_allow = [\"192.0.2.10/32\"]\n[profiles.agent]\nimage = \"example.invalid/image:v1\"\n[profiles.agent.readiness]\ndenied = [{{ address = \"10.0.0.1\", port = 443 }}]\npermitted = [{{ address = \"192.0.2.10\", port = 443 }}]\ndns = [{{ resolver = \"198.51.100.53\", query = \"example.invalid\" }}]\n",
|
|
root.join("data").display()
|
|
),
|
|
)
|
|
.unwrap();
|
|
load_trusted_configuration(ConfigurationLoadRequest {
|
|
explicit_configuration_path: Some(configuration_path),
|
|
selected_profile: "agent".into(),
|
|
workspace,
|
|
home: root.join("home"),
|
|
xdg_config_home: None,
|
|
xdg_data_home: None,
|
|
xdg_runtime_dir: runtime,
|
|
environment: BTreeMap::new(),
|
|
sandbox_layout: default_sandbox_layout().unwrap(),
|
|
})
|
|
.unwrap()
|
|
}
|
|
|
|
fn generation(configuration: &ResolvedDomainConfiguration) -> Generation {
|
|
Generation {
|
|
domain_id: configuration.domain_id(),
|
|
generation_id: GenerationId::new(1),
|
|
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,
|
|
}
|
|
}
|
|
|
|
fn session(seed: u8) -> Session {
|
|
Session {
|
|
generation_id: GenerationId::new(1),
|
|
session_id: SessionId::new([seed; 32]),
|
|
cli_id: ClientId::new([seed + 32; 32]),
|
|
session_anchor_id: SessionAnchorId::new([seed + 64; 32]),
|
|
session_phase: SessionPhase::New,
|
|
}
|
|
}
|
|
}
|