Bind cold start to production authorities
This commit is contained in:
parent
a9c9cf7ab3
commit
b2f178d25e
8 changed files with 816 additions and 30 deletions
|
|
@ -40,5 +40,9 @@ name = "escalation"
|
|||
path = "tests/escalation.rs"
|
||||
harness = false
|
||||
|
||||
[[test]]
|
||||
name = "production"
|
||||
path = "tests/production.rs"
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
|
|
|
|||
|
|
@ -6,5 +6,6 @@ pub mod domain_store;
|
|||
pub mod escalation;
|
||||
pub mod lifecycle;
|
||||
pub mod network;
|
||||
pub mod production;
|
||||
pub mod runtime;
|
||||
pub mod state_anchor;
|
||||
|
|
|
|||
|
|
@ -282,17 +282,53 @@ where
|
|||
generation: Generation,
|
||||
session: Session,
|
||||
) -> Result<(), LifecycleError<A::Error>> {
|
||||
self.bootstrap_new_domain_while_owner_live(generation, session, || true)
|
||||
.map(|_| ())
|
||||
}
|
||||
|
||||
/// Drives cold start while checking the bootstrap owner's exact lifetime authority.
|
||||
///
|
||||
/// Returns `false` after applying the modeled bootstrap-owner disconnect transition.
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns the first invalid transition, durable update, or production adapter failure.
|
||||
pub fn bootstrap_new_domain_while_owner_live(
|
||||
&mut self,
|
||||
generation: Generation,
|
||||
session: Session,
|
||||
mut owner_live: impl FnMut() -> bool,
|
||||
) -> Result<bool, LifecycleError<A::Error>> {
|
||||
let session_id = session.session_id;
|
||||
self.start_new_domain(generation, session)?;
|
||||
macro_rules! checkpoint {
|
||||
() => {
|
||||
if !owner_live() {
|
||||
self.disconnect_session(session_id)?;
|
||||
return Ok(false);
|
||||
}
|
||||
};
|
||||
}
|
||||
checkpoint!();
|
||||
self.create_trust_anchor()?;
|
||||
checkpoint!();
|
||||
self.start_runtime_supervisor()?;
|
||||
checkpoint!();
|
||||
self.lease_prepared_lower()?;
|
||||
checkpoint!();
|
||||
self.mount_state_anchor()?;
|
||||
checkpoint!();
|
||||
self.start_sidecar()?;
|
||||
checkpoint!();
|
||||
self.start_connectivity_helper()?;
|
||||
checkpoint!();
|
||||
self.mark_network_ready()?;
|
||||
checkpoint!();
|
||||
self.arm_runtime_causal_edge()?;
|
||||
checkpoint!();
|
||||
self.start_sandbox()?;
|
||||
self.publish_domain()
|
||||
checkpoint!();
|
||||
self.publish_domain()?;
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
/// # Errors
|
||||
|
|
|
|||
445
crates/lazaret-host-coordinator/src/production.rs
Normal file
445
crates/lazaret-host-coordinator/src/production.rs
Normal file
|
|
@ -0,0 +1,445 @@
|
|||
use std::ffi::OsString;
|
||||
use std::fmt;
|
||||
use std::fs::{self, File, OpenOptions};
|
||||
use std::io::Write;
|
||||
use std::os::fd::{AsFd, OwnedFd};
|
||||
use std::os::unix::ffi::OsStringExt;
|
||||
use std::os::unix::fs::{FileExt, PermissionsExt};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use lazaret_core::configuration::ResolvedDomainConfiguration;
|
||||
use lazaret_core::identity::{BundleDigest, ContainerId};
|
||||
use lazaret_core::network::{ReachabilityExpectation, derive_ipv4_policy};
|
||||
use lazaret_core::records::{
|
||||
Generation, GenerationIdentity, HostProcessOwnershipSlot, IpAddress, PreparedArtifact, Session,
|
||||
};
|
||||
use lazaret_linux::filesystem::namespace_identity;
|
||||
use lazaret_linux::process::ProcessOwnershipContext;
|
||||
use lazaret_runtime_supervisor::network::NetworkProbeRequest;
|
||||
use lazaret_runtime_supervisor::sandbox::{
|
||||
RootProcess, SandboxBuildRequest, SandboxBundle, build_sandbox_bundle,
|
||||
};
|
||||
use lazaret_runtime_supervisor::supervision::{RuntimeNetworkLaunch, RuntimeSupervisorLaunch};
|
||||
use lazaret_state_anchor::{StateAnchorLaunchAuthority, StateAnchorLayout};
|
||||
use lazaret_storage_catalog::Catalog;
|
||||
use rustix::event::{PollFd, PollFlags, Timespec, poll};
|
||||
use rustix::fs::{FileType, fstat};
|
||||
use rustix::io::{FdFlags, fcntl_setfd};
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
use crate::bootstrap::{
|
||||
ProductionBootstrapPlan, ProductionHostCoordinatorLifecycle, production_lifecycle,
|
||||
};
|
||||
use crate::domain_store::HostCoordinatorDomainStore;
|
||||
|
||||
const NETWORK_NAMESPACE_PATH: &str = "/proc/self/ns/net";
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct ProductionColdStartError(String);
|
||||
|
||||
impl fmt::Display for ProductionColdStartError {
|
||||
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
formatter.write_str(&self.0)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for ProductionColdStartError {}
|
||||
|
||||
fn failure(error: impl fmt::Display) -> ProductionColdStartError {
|
||||
ProductionColdStartError(error.to_string())
|
||||
}
|
||||
|
||||
pub struct ProductionColdStartInputs<'a> {
|
||||
pub store: Arc<HostCoordinatorDomainStore>,
|
||||
pub catalog: &'a Catalog,
|
||||
pub configuration: &'a ResolvedDomainConfiguration,
|
||||
pub generation: Generation,
|
||||
pub prepared_artifact: &'a PreparedArtifact,
|
||||
pub root_process: RootProcess,
|
||||
pub resolver_addresses: Vec<IpAddress>,
|
||||
pub network_probe: NetworkProbeRequest,
|
||||
pub reachability: ReachabilityExpectation,
|
||||
pub container_id: ContainerId,
|
||||
pub invoking_uid: u32,
|
||||
pub escalation_client: &'a File,
|
||||
}
|
||||
|
||||
pub struct BootstrapOwnerGuard {
|
||||
descriptor: OwnedFd,
|
||||
}
|
||||
|
||||
impl BootstrapOwnerGuard {
|
||||
/// Adopts an inherited bootstrap-owner lifetime pipe without PID sampling.
|
||||
///
|
||||
/// # Errors
|
||||
/// Refuses a descriptor that is not a pipe or cannot be confined to this process.
|
||||
pub fn from_inherited(descriptor: OwnedFd) -> Result<Self, ProductionColdStartError> {
|
||||
let metadata = fstat(&descriptor).map_err(failure)?;
|
||||
if !FileType::from_raw_mode(metadata.st_mode).is_fifo() {
|
||||
return Err(ProductionColdStartError(
|
||||
"bootstrap-owner lifetime descriptor is not a pipe".into(),
|
||||
));
|
||||
}
|
||||
fcntl_setfd(descriptor.as_fd(), FdFlags::CLOEXEC).map_err(failure)?;
|
||||
Ok(Self { descriptor })
|
||||
}
|
||||
|
||||
fn is_live(&self) -> std::io::Result<bool> {
|
||||
let mut descriptors = [PollFd::new(&self.descriptor, PollFlags::IN)];
|
||||
poll(&mut descriptors, Some(&Timespec::default()))?;
|
||||
Ok(!descriptors[0]
|
||||
.revents()
|
||||
.intersects(PollFlags::IN | PollFlags::HUP | PollFlags::ERR))
|
||||
}
|
||||
}
|
||||
|
||||
pub struct ProductionColdStart<'a> {
|
||||
store: Arc<HostCoordinatorDomainStore>,
|
||||
catalog: &'a Catalog,
|
||||
configuration: &'a ResolvedDomainConfiguration,
|
||||
generation: Generation,
|
||||
generation_identity: GenerationIdentity,
|
||||
prepared_artifact: &'a PreparedArtifact,
|
||||
state_anchor_layout: StateAnchorLayout<'a>,
|
||||
runtime_launch: Option<RuntimeSupervisorLaunch>,
|
||||
sandbox_bundle: SandboxBundle,
|
||||
network_policy: lazaret_core::network::CanonicalDerivedNetworkPolicy,
|
||||
network_probe: NetworkProbeRequest,
|
||||
reachability: ReachabilityExpectation,
|
||||
container_id: ContainerId,
|
||||
runtime_root: PathBuf,
|
||||
bundle_digest: BundleDigest,
|
||||
invoking_uid: u32,
|
||||
_staging: RuntimeStaging,
|
||||
}
|
||||
|
||||
impl<'a> ProductionColdStart<'a> {
|
||||
/// Stages one production generation from validated component authorities.
|
||||
///
|
||||
/// # Errors
|
||||
/// Refuses mismatched generation/configuration/probe inputs, changed host authority, unsafe
|
||||
/// staging paths, or component construction failure.
|
||||
pub fn stage(inputs: ProductionColdStartInputs<'a>) -> Result<Self, ProductionColdStartError> {
|
||||
require_inputs(&inputs)?;
|
||||
let runtime_domain = host_path(
|
||||
inputs
|
||||
.configuration
|
||||
.runtime_domain_root()
|
||||
.canonical_label()
|
||||
.as_bytes(),
|
||||
);
|
||||
let state_anchor_layout = StateAnchorLayout::create(inputs.configuration)
|
||||
.map_err(|error| failure(format!("State Anchor layout: {error}")))?;
|
||||
let staging = RuntimeStaging::create(
|
||||
&runtime_domain,
|
||||
inputs.container_id,
|
||||
inputs.escalation_client,
|
||||
)
|
||||
.map_err(|error| failure(format!("runtime staging: {error}")))?;
|
||||
let expected_network_namespace = namespace_identity(Path::new(NETWORK_NAMESPACE_PATH))
|
||||
.map_err(|error| failure(format!("network namespace: {error}")))?;
|
||||
let sandbox_bundle = build_sandbox_bundle(SandboxBuildRequest {
|
||||
configuration: inputs.configuration,
|
||||
container_id: inputs.container_id,
|
||||
expected_network_namespace,
|
||||
root_process: inputs.root_process,
|
||||
terminal: None,
|
||||
})
|
||||
.map_err(|error| failure(format!("sandbox bundle: {error}")))?;
|
||||
staging
|
||||
.write_bundle(sandbox_bundle.specification_json())
|
||||
.map_err(|error| failure(format!("bundle staging: {error}")))?;
|
||||
let network_policy = derive_ipv4_policy(inputs.configuration, inputs.resolver_addresses)
|
||||
.map_err(|error| failure(format!("network policy: {error}")))?;
|
||||
let ownership = ProcessOwnershipContext {
|
||||
domain_id: inputs.generation.domain_id,
|
||||
generation_id: inputs.generation.generation_id,
|
||||
coordinator_id: inputs.generation.coordinator_id,
|
||||
slot: HostProcessOwnershipSlot::SandboxRuntimeSupervisor,
|
||||
};
|
||||
let state_anchor =
|
||||
StateAnchorLaunchAuthority::capture(&state_anchor_layout, inputs.prepared_artifact)
|
||||
.map_err(|error| failure(format!("State Anchor authority: {error}")))?;
|
||||
let runtime_launch = RuntimeSupervisorLaunch::from_networked_runsc_command(
|
||||
ownership,
|
||||
sandbox_bundle.command(),
|
||||
RuntimeNetworkLaunch {
|
||||
policy: &network_policy,
|
||||
configuration_fingerprint: inputs.generation.configuration_fingerprint,
|
||||
network_sidecar_id: inputs.generation.network_sidecar_id,
|
||||
},
|
||||
)
|
||||
.map_err(|error| failure(format!("Runtime Supervisor launch: {error}")))?
|
||||
.with_state_anchor(state_anchor);
|
||||
let bundle_digest =
|
||||
BundleDigest::new(Sha256::digest(sandbox_bundle.specification_json()).into());
|
||||
let generation_identity = GenerationIdentity {
|
||||
agent_profile: inputs.configuration.selected_profile().to_owned(),
|
||||
domain_id: inputs.generation.domain_id,
|
||||
generation_id: inputs.generation.generation_id,
|
||||
configuration_fingerprint: inputs.generation.configuration_fingerprint,
|
||||
coordinator_id: inputs.generation.coordinator_id,
|
||||
sandbox_id: inputs.generation.sandbox_id,
|
||||
};
|
||||
Ok(Self {
|
||||
store: inputs.store,
|
||||
catalog: inputs.catalog,
|
||||
configuration: inputs.configuration,
|
||||
generation: inputs.generation,
|
||||
generation_identity,
|
||||
prepared_artifact: inputs.prepared_artifact,
|
||||
state_anchor_layout,
|
||||
runtime_launch: Some(runtime_launch),
|
||||
sandbox_bundle,
|
||||
network_policy,
|
||||
network_probe: inputs.network_probe,
|
||||
reachability: inputs.reachability,
|
||||
container_id: inputs.container_id,
|
||||
runtime_root: runtime_domain.join("runsc"),
|
||||
bundle_digest,
|
||||
invoking_uid: inputs.invoking_uid,
|
||||
_staging: staging,
|
||||
})
|
||||
}
|
||||
|
||||
/// Runs cold start and exposes the exact retained lifecycle to one production service loop.
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns staging, lifecycle, or service-loop failure. The service is single-use.
|
||||
pub fn run<Service>(
|
||||
mut self,
|
||||
bootstrap_owner: Session,
|
||||
owner_guard: &BootstrapOwnerGuard,
|
||||
service: Service,
|
||||
) -> Result<(), ProductionColdStartError>
|
||||
where
|
||||
Service: FnOnce(
|
||||
&Mutex<ProductionHostCoordinatorLifecycle<'_, '_>>,
|
||||
) -> Result<(), ProductionColdStartError>,
|
||||
{
|
||||
let store = Arc::clone(&self.store);
|
||||
let generation = self.generation.clone();
|
||||
let plan = self.take_plan()?;
|
||||
let mut lifecycle = production_lifecycle(&store, plan).map_err(failure)?;
|
||||
let published = lifecycle
|
||||
.bootstrap_new_domain_while_owner_live(generation, bootstrap_owner, || {
|
||||
owner_guard.is_live().unwrap_or(false)
|
||||
})
|
||||
.map_err(failure)?;
|
||||
if !published {
|
||||
return Err(ProductionColdStartError(
|
||||
"bootstrap owner disconnected before publication".into(),
|
||||
));
|
||||
}
|
||||
service(&Mutex::new(lifecycle))
|
||||
}
|
||||
|
||||
fn take_plan(&mut self) -> Result<ProductionBootstrapPlan<'_>, ProductionColdStartError> {
|
||||
Ok(ProductionBootstrapPlan {
|
||||
store: Arc::clone(&self.store),
|
||||
catalog: self.catalog,
|
||||
configuration: self.configuration,
|
||||
generation_identity: self.generation_identity.clone(),
|
||||
prepared_artifact: self.prepared_artifact,
|
||||
state_anchor_layout: &self.state_anchor_layout,
|
||||
runtime_launch: self.runtime_launch.take().ok_or_else(|| {
|
||||
ProductionColdStartError("production cold start was already consumed".into())
|
||||
})?,
|
||||
sandbox_bundle: &self.sandbox_bundle,
|
||||
network_policy: &self.network_policy,
|
||||
network_probe: self.network_probe.clone(),
|
||||
reachability: &self.reachability,
|
||||
container_id: self.container_id,
|
||||
runtime_root: self.runtime_root.clone(),
|
||||
bundle_digest: self.bundle_digest,
|
||||
invoking_uid: self.invoking_uid,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
fn require_inputs(inputs: &ProductionColdStartInputs<'_>) -> Result<(), ProductionColdStartError> {
|
||||
if inputs.generation.domain_id != inputs.configuration.domain_id()
|
||||
|| inputs.generation.configuration_fingerprint
|
||||
!= inputs.configuration.configuration_fingerprint()
|
||||
|| inputs.network_probe.domain_id != inputs.generation.domain_id
|
||||
|| inputs.network_probe.generation_id != inputs.generation.generation_id
|
||||
|| inputs.resolver_addresses.is_empty()
|
||||
{
|
||||
return Err(ProductionColdStartError(
|
||||
"production cold-start inputs disagree".into(),
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
struct RuntimeStaging {
|
||||
bundle: PathBuf,
|
||||
control: PathBuf,
|
||||
}
|
||||
|
||||
impl RuntimeStaging {
|
||||
fn create(
|
||||
runtime_domain: &Path,
|
||||
container_id: ContainerId,
|
||||
escalation_client: &File,
|
||||
) -> Result<Self, ProductionColdStartError> {
|
||||
let name = hex(container_id.as_bytes());
|
||||
let bundles = runtime_domain.join("bundles");
|
||||
let controls = runtime_domain.join("sandbox-control");
|
||||
let runsc = runtime_domain.join("runsc");
|
||||
for directory in [&bundles, &controls, &runsc] {
|
||||
ensure_directory(directory)
|
||||
.map_err(|error| failure(format!("creating {}: {error}", directory.display())))?;
|
||||
}
|
||||
let bundle = bundles.join(&name);
|
||||
let control = controls.join(name);
|
||||
fs::create_dir(&bundle)
|
||||
.map_err(|error| failure(format!("creating bundle leaf: {error}")))?;
|
||||
set_private_directory(&bundle)?;
|
||||
if let Err(error) = fs::create_dir(&control) {
|
||||
let _ = fs::remove_dir(&bundle);
|
||||
return Err(failure(error));
|
||||
}
|
||||
if let Err(error) = set_private_directory(&control) {
|
||||
let _ = fs::remove_dir(&control);
|
||||
let _ = fs::remove_dir(&bundle);
|
||||
return Err(error);
|
||||
}
|
||||
let staging = Self { bundle, control };
|
||||
staging.populate_control(escalation_client)?;
|
||||
Ok(staging)
|
||||
}
|
||||
|
||||
fn populate_control(&self, escalation_client: &File) -> Result<(), ProductionColdStartError> {
|
||||
for relative in [
|
||||
"credentials",
|
||||
"mailbox",
|
||||
"mailbox/requests",
|
||||
"mailbox/results",
|
||||
"bin",
|
||||
] {
|
||||
let directory = self.control.join(relative);
|
||||
fs::create_dir(&directory)
|
||||
.map_err(|error| failure(format!("creating control {relative}: {error}")))?;
|
||||
set_private_directory(&directory)?;
|
||||
}
|
||||
let source = escalation_client
|
||||
.metadata()
|
||||
.map_err(|error| failure(format!("inspecting escalation client: {error}")))?;
|
||||
if !source.file_type().is_file() || source.permissions().mode() & 0o111 == 0 {
|
||||
return Err(ProductionColdStartError(
|
||||
"escalation client is not an executable regular file".into(),
|
||||
));
|
||||
}
|
||||
copy_descriptor(
|
||||
escalation_client,
|
||||
&self.control.join("bin/lazaret-escalate"),
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn write_bundle(&self, bytes: &[u8]) -> Result<(), ProductionColdStartError> {
|
||||
let path = self.bundle.join("config.json");
|
||||
let mut file = OpenOptions::new()
|
||||
.write(true)
|
||||
.create_new(true)
|
||||
.open(path)
|
||||
.map_err(failure)?;
|
||||
file.write_all(bytes).map_err(failure)?;
|
||||
file.set_permissions(fs::Permissions::from_mode(0o600))
|
||||
.map_err(failure)?;
|
||||
file.sync_all().map_err(failure)?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn copy_descriptor(source: &File, destination: &Path) -> Result<(), ProductionColdStartError> {
|
||||
let mut destination = OpenOptions::new()
|
||||
.write(true)
|
||||
.create_new(true)
|
||||
.open(destination)
|
||||
.map_err(failure)?;
|
||||
destination
|
||||
.set_permissions(fs::Permissions::from_mode(0o700))
|
||||
.map_err(failure)?;
|
||||
let mut offset = 0_u64;
|
||||
let mut buffer = [0_u8; 16 * 1024];
|
||||
loop {
|
||||
let read = source.read_at(&mut buffer, offset).map_err(failure)?;
|
||||
if read == 0 {
|
||||
break;
|
||||
}
|
||||
destination.write_all(&buffer[..read]).map_err(failure)?;
|
||||
offset = offset
|
||||
.checked_add(read as u64)
|
||||
.ok_or_else(|| ProductionColdStartError("escalation client is too large".into()))?;
|
||||
}
|
||||
destination.sync_all().map_err(failure)
|
||||
}
|
||||
|
||||
fn set_private_directory(path: &Path) -> Result<(), ProductionColdStartError> {
|
||||
fs::set_permissions(path, fs::Permissions::from_mode(0o700)).map_err(failure)
|
||||
}
|
||||
|
||||
impl Drop for RuntimeStaging {
|
||||
fn drop(&mut self) {
|
||||
let _ = fs::remove_dir_all(&self.control);
|
||||
let _ = fs::remove_dir_all(&self.bundle);
|
||||
}
|
||||
}
|
||||
|
||||
fn ensure_directory(path: &Path) -> Result<(), ProductionColdStartError> {
|
||||
match fs::create_dir(path) {
|
||||
Ok(()) => Ok(()),
|
||||
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
|
||||
let metadata = fs::symlink_metadata(path).map_err(failure)?;
|
||||
if metadata.file_type().is_dir() {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(ProductionColdStartError(
|
||||
"production staging path is not a directory".into(),
|
||||
))
|
||||
}
|
||||
}
|
||||
Err(error) => Err(failure(error)),
|
||||
}
|
||||
}
|
||||
|
||||
fn host_path(bytes: &[u8]) -> PathBuf {
|
||||
PathBuf::from(OsString::from_vec(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}").expect("writing to a String cannot fail");
|
||||
output
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::fs::File;
|
||||
use std::os::fd::OwnedFd;
|
||||
|
||||
use rustix::pipe::{PipeFlags, pipe_with};
|
||||
|
||||
use super::BootstrapOwnerGuard;
|
||||
|
||||
#[test]
|
||||
fn bootstrap_owner_guard_tracks_the_exact_inherited_pipe() {
|
||||
let (reader, writer) = pipe_with(PipeFlags::CLOEXEC).unwrap();
|
||||
let guard = BootstrapOwnerGuard::from_inherited(reader).unwrap();
|
||||
assert!(guard.is_live().unwrap());
|
||||
drop(writer);
|
||||
assert!(!guard.is_live().unwrap());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bootstrap_owner_guard_refuses_non_pipe_authority() {
|
||||
let descriptor: OwnedFd = File::open("/dev/null").unwrap().into();
|
||||
assert!(BootstrapOwnerGuard::from_inherited(descriptor).is_err());
|
||||
}
|
||||
}
|
||||
|
|
@ -1,3 +1,4 @@
|
|||
use std::cell::Cell;
|
||||
use std::fs::OpenOptions;
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
|
|
@ -225,6 +226,36 @@ fn complete_startup(
|
|||
lifecycle.publish_domain().unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bootstrap_owner_guard_failure_uses_exact_modeled_disconnect_transition() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let store = open_store(&directory);
|
||||
let (domain_adapter, session_adapter) = adapters();
|
||||
let actions = Arc::clone(&domain_adapter.actions);
|
||||
let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter);
|
||||
let checks = Cell::new(0_u8);
|
||||
let owner = session(10, 1);
|
||||
let published = lifecycle
|
||||
.bootstrap_new_domain_while_owner_live(generation(1), owner.clone(), || {
|
||||
checks.set(checks.get() + 1);
|
||||
checks.get() < 3
|
||||
})
|
||||
.unwrap();
|
||||
assert!(!published);
|
||||
assert_eq!(lifecycle.phase(), DomainPhase::Failed);
|
||||
assert!(!lifecycle.is_published());
|
||||
assert_eq!(
|
||||
lifecycle.session_phase(owner.session_id),
|
||||
Some(SessionPhase::Closed)
|
||||
);
|
||||
assert!(
|
||||
!actions
|
||||
.lock()
|
||||
.unwrap()
|
||||
.contains(&LifecycleAction::PublishDomain)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn exact_start_join_disconnect_and_teardown_actions_are_ordered() {
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
|
|
|
|||
185
crates/lazaret-host-coordinator/tests/production.rs
Normal file
185
crates/lazaret-host-coordinator/tests/production.rs
Normal file
|
|
@ -0,0 +1,185 @@
|
|||
use std::collections::BTreeMap;
|
||||
use std::fs::{self, File, OpenOptions};
|
||||
use std::os::unix::ffi::{OsStrExt, OsStringExt};
|
||||
use std::os::unix::fs::{MetadataExt, PermissionsExt};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
|
||||
use lazaret_core::configuration::{ConfigurationLoadRequest, load_trusted_configuration};
|
||||
use lazaret_core::identity::{
|
||||
ContainerId, CoordinatorId, DeviceId, GenerationId, HostPath, IdMapDigest, InodeId, PolicyId,
|
||||
PreparedArtifactId, SandboxId, SelectedManifestDigest, SourceIdentity,
|
||||
};
|
||||
use lazaret_core::records::{
|
||||
ArtifactState, Domain, DurableDecision, Generation, IpAddress, PreparedArtifact,
|
||||
};
|
||||
use lazaret_host_coordinator::domain_store::HostCoordinatorDomainStore;
|
||||
use lazaret_host_coordinator::production::{ProductionColdStart, ProductionColdStartInputs};
|
||||
use lazaret_linux::filesystem::{namespace_identity, try_exclusive_flock};
|
||||
use lazaret_runtime_supervisor::network::NetworkProbeRequest;
|
||||
use lazaret_runtime_supervisor::sandbox::{RootProcess, production_sandbox_layout};
|
||||
use lazaret_storage_catalog::Catalog;
|
||||
use rustix::process::getuid;
|
||||
|
||||
#[test]
|
||||
fn validated_inputs_stage_one_complete_plan_and_drop_exact_runtime_artifacts() {
|
||||
let tree = tempfile::tempdir().unwrap();
|
||||
let configuration = configuration(tree.path());
|
||||
let persistent_domain = host_path(
|
||||
configuration
|
||||
.persistent_domain_root()
|
||||
.canonical_label()
|
||||
.as_bytes(),
|
||||
);
|
||||
let runtime_domain = host_path(
|
||||
configuration
|
||||
.runtime_domain_root()
|
||||
.canonical_label()
|
||||
.as_bytes(),
|
||||
);
|
||||
let catalog = Catalog::new(
|
||||
tree.path().join("catalog.redb"),
|
||||
tree.path().join("catalog.lock"),
|
||||
);
|
||||
catalog.initialize().unwrap();
|
||||
let domain = Domain {
|
||||
domain_id: configuration.domain_id(),
|
||||
profile_name: configuration.selected_profile().to_owned(),
|
||||
canonical_workspace: configuration.workspace().canonical_label().clone(),
|
||||
};
|
||||
let lock = OpenOptions::new()
|
||||
.read(true)
|
||||
.write(true)
|
||||
.create_new(true)
|
||||
.open(persistent_domain.join("domain.lock"))
|
||||
.unwrap();
|
||||
let store = Arc::new(
|
||||
HostCoordinatorDomainStore::open(
|
||||
try_exclusive_flock(lock).unwrap(),
|
||||
&persistent_domain.join("domain.redb"),
|
||||
&domain,
|
||||
)
|
||||
.unwrap(),
|
||||
);
|
||||
let lower = tree.path().join("prepared-lower");
|
||||
fs::create_dir(&lower).unwrap();
|
||||
let metadata = lower.metadata().unwrap();
|
||||
let artifact = PreparedArtifact {
|
||||
prepared_artifact_id: PreparedArtifactId::new([8; 32]),
|
||||
source_identity: SourceIdentity::new(b"synthetic-spike".to_vec()),
|
||||
selected_manifest_digest: SelectedManifestDigest::new([9; 32]),
|
||||
policy_id: PolicyId::new([10; 32]),
|
||||
id_map_digest: IdMapDigest::new([11; 32]),
|
||||
path: HostPath::new(lower.as_os_str().as_bytes().to_vec()),
|
||||
device: DeviceId::new(metadata.dev()),
|
||||
inode: InodeId::new(metadata.ino()),
|
||||
state: ArtifactState::Available,
|
||||
};
|
||||
let generation = Generation {
|
||||
domain_id: configuration.domain_id(),
|
||||
generation_id: GenerationId::new(1),
|
||||
configuration_fingerprint: configuration.configuration_fingerprint(),
|
||||
coordinator_id: CoordinatorId::new([2; 32]),
|
||||
network_sidecar_id: lazaret_core::identity::NetworkSidecarId::new([3; 32]),
|
||||
sandbox_id: SandboxId::new([4; 32]),
|
||||
durable_decision: DurableDecision::None,
|
||||
};
|
||||
let container_id = ContainerId::new([5; 32]);
|
||||
let escalation_client = File::open("/bin/true").unwrap();
|
||||
let invalid_escalation_client = File::open("/dev/null").unwrap();
|
||||
let inputs = |escalation_client| ProductionColdStartInputs {
|
||||
store: Arc::clone(&store),
|
||||
catalog: &catalog,
|
||||
configuration: &configuration,
|
||||
generation: generation.clone(),
|
||||
prepared_artifact: &artifact,
|
||||
root_process: RootProcess::new("/bin/sh", ["-c".into(), "sleep infinity".into()]).unwrap(),
|
||||
resolver_addresses: vec![IpAddress::V4([10, 0, 2, 3])],
|
||||
network_probe: NetworkProbeRequest {
|
||||
domain_id: configuration.domain_id(),
|
||||
generation_id: GenerationId::new(1),
|
||||
enforcement_namespace: namespace_identity(Path::new("/proc/self/ns/net")).unwrap(),
|
||||
denied: Vec::new(),
|
||||
permitted: Vec::new(),
|
||||
dns: Vec::new(),
|
||||
},
|
||||
reachability: lazaret_core::network::ReachabilityExpectation {
|
||||
denied_destinations: Vec::new(),
|
||||
permitted_destinations: Vec::new(),
|
||||
dns_queries: Vec::new(),
|
||||
},
|
||||
container_id,
|
||||
invoking_uid: getuid().as_raw(),
|
||||
escalation_client,
|
||||
};
|
||||
let staged = ProductionColdStart::stage(inputs(&escalation_client)).unwrap();
|
||||
let name = hex(container_id.as_bytes());
|
||||
let bundle = runtime_domain.join("bundles").join(&name);
|
||||
let control = runtime_domain.join("sandbox-control").join(&name);
|
||||
assert_private_artifacts(&bundle, &control);
|
||||
drop(staged);
|
||||
assert!(!bundle.exists());
|
||||
assert!(!control.exists());
|
||||
assert!(ProductionColdStart::stage(inputs(&invalid_escalation_client)).is_err());
|
||||
assert!(!bundle.exists());
|
||||
assert!(!control.exists());
|
||||
}
|
||||
|
||||
fn assert_private_artifacts(bundle: &Path, control: &Path) {
|
||||
let config = bundle.join("config.json");
|
||||
let escalation_client = control.join("bin/lazaret-escalate");
|
||||
assert!(config.is_file());
|
||||
assert!(escalation_client.is_file());
|
||||
assert!(control.join("credentials").is_dir());
|
||||
assert_eq!(mode(bundle), 0o700);
|
||||
assert_eq!(mode(control), 0o700);
|
||||
assert_eq!(mode(&config), 0o600);
|
||||
assert_eq!(mode(&escalation_client), 0o700);
|
||||
}
|
||||
|
||||
fn mode(path: &Path) -> u32 {
|
||||
path.metadata().unwrap().permissions().mode() & 0o777
|
||||
}
|
||||
|
||||
fn configuration(root: &Path) -> lazaret_core::configuration::ResolvedDomainConfiguration {
|
||||
let configuration_directory = root.join("configuration");
|
||||
let workspace = root.join("workspace");
|
||||
let runtime = root.join("runtime");
|
||||
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\n[storage]\ndata_root = \"{}\"\n\n[profiles.default]\nimage = \"example.invalid/factory:v1\"\npersistent_dirs = []\n",
|
||||
root.join("data").display()
|
||||
),
|
||||
)
|
||||
.unwrap();
|
||||
load_trusted_configuration(ConfigurationLoadRequest {
|
||||
explicit_configuration_path: Some(configuration_path),
|
||||
selected_profile: "default".into(),
|
||||
workspace,
|
||||
home: root.join("home"),
|
||||
xdg_config_home: None,
|
||||
xdg_data_home: None,
|
||||
xdg_runtime_dir: runtime,
|
||||
environment: BTreeMap::new(),
|
||||
sandbox_layout: production_sandbox_layout().unwrap(),
|
||||
})
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
fn host_path(bytes: &[u8]) -> PathBuf {
|
||||
PathBuf::from(std::ffi::OsString::from_vec(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
|
||||
})
|
||||
}
|
||||
|
|
@ -328,6 +328,40 @@ impl ChildBroker {
|
|||
}
|
||||
}
|
||||
|
||||
/// Starts a non-durable direct child with split captured output and exact retained authority.
|
||||
///
|
||||
/// # Errors
|
||||
/// Refuses invalid parent-edge or SIGCHLD state and cleans up on spawn, pidfd, barrier, or
|
||||
/// pipe-acquisition failure.
|
||||
#[allow(clippy::disallowed_methods, clippy::disallowed_types)]
|
||||
pub fn run_transient_captured(&self, target: &TargetCommand) -> io::Result<CapturedProcess> {
|
||||
let _guard = self
|
||||
.exclusion
|
||||
.lock()
|
||||
.map_err(|_| io::Error::other("child broker lock poisoned"))?;
|
||||
require_captured_parent_edge(target.parent_edge())?;
|
||||
require_default_sigchld()?;
|
||||
require_reapable_sigchld()?;
|
||||
let parent = current_process_identity()?;
|
||||
let (release, wait) = launch_barrier()?;
|
||||
let mut command = private_role_command(WRAPPER_ROLE);
|
||||
append_parent_identity(&mut command, parent);
|
||||
target.append_to(&mut command);
|
||||
command
|
||||
.stdin(wait.into_stdio())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::piped());
|
||||
let mut child = command.spawn()?;
|
||||
let process = match acquire_spawned_child(child.id()) {
|
||||
Ok(process) => process,
|
||||
Err(error) => return fail_before_pidfd(release, &mut child, error),
|
||||
};
|
||||
if let Err(error) = release.release() {
|
||||
return fail_after_pidfd_closed(process, error);
|
||||
}
|
||||
captured_process(process, child)
|
||||
}
|
||||
|
||||
/// Spawns the repository wrapper, stages exact ownership, accepts its acknowledgment,
|
||||
/// and only then releases target execution.
|
||||
///
|
||||
|
|
@ -530,34 +564,7 @@ impl ChildBroker {
|
|||
if let Err(error) = release.release() {
|
||||
return fail_after_pidfd_closed(process, error);
|
||||
}
|
||||
let Some(stdout) = child.stdout.take() else {
|
||||
return fail_running_captured(
|
||||
process,
|
||||
child,
|
||||
io::Error::other("captured child stdout is unavailable"),
|
||||
);
|
||||
};
|
||||
let Some(stderr) = child.stderr.take() else {
|
||||
return fail_running_captured(
|
||||
process,
|
||||
child,
|
||||
io::Error::other("captured child stderr is unavailable"),
|
||||
);
|
||||
};
|
||||
let stdout = match child_pipe_file(stdout.into()) {
|
||||
Ok(stdout) => stdout,
|
||||
Err(error) => return fail_running_captured(process, child, error),
|
||||
};
|
||||
let stderr = match child_pipe_file(stderr.into()) {
|
||||
Ok(stderr) => stderr,
|
||||
Err(error) => return fail_running_captured(process, child, error),
|
||||
};
|
||||
drop(child);
|
||||
Ok(CapturedProcess {
|
||||
process,
|
||||
stdout,
|
||||
stderr,
|
||||
})
|
||||
captured_process(process, child)
|
||||
}
|
||||
|
||||
/// # Errors
|
||||
|
|
@ -587,6 +594,41 @@ impl ChildBroker {
|
|||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::disallowed_types)]
|
||||
fn captured_process(
|
||||
process: RetainedProcess,
|
||||
mut child: std::process::Child,
|
||||
) -> io::Result<CapturedProcess> {
|
||||
let Some(stdout) = child.stdout.take() else {
|
||||
return fail_running_captured(
|
||||
process,
|
||||
child,
|
||||
io::Error::other("captured child stdout is unavailable"),
|
||||
);
|
||||
};
|
||||
let Some(stderr) = child.stderr.take() else {
|
||||
return fail_running_captured(
|
||||
process,
|
||||
child,
|
||||
io::Error::other("captured child stderr is unavailable"),
|
||||
);
|
||||
};
|
||||
let stdout = match child_pipe_file(stdout.into()) {
|
||||
Ok(stdout) => stdout,
|
||||
Err(error) => return fail_running_captured(process, child, error),
|
||||
};
|
||||
let stderr = match child_pipe_file(stderr.into()) {
|
||||
Ok(stderr) => stderr,
|
||||
Err(error) => return fail_running_captured(process, child, error),
|
||||
};
|
||||
drop(child);
|
||||
Ok(CapturedProcess {
|
||||
process,
|
||||
stdout,
|
||||
stderr,
|
||||
})
|
||||
}
|
||||
|
||||
fn child_pipe_file(fd: OwnedFd) -> io::Result<File> {
|
||||
let flags = fcntl_getfl(fd.as_fd())?;
|
||||
fcntl_setfl(fd.as_fd(), flags | OFlags::NONBLOCK)?;
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
use std::ffi::OsString;
|
||||
use std::fs::File;
|
||||
use std::io::{Seek, Write};
|
||||
use std::io::{Read, Seek, Write};
|
||||
use std::os::fd::AsFd;
|
||||
use std::process::{ExitCode, Stdio};
|
||||
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
|
||||
|
|
@ -45,6 +45,7 @@ fn exit(result: std::io::Result<()>) -> ExitCode {
|
|||
|
||||
fn run_all() -> std::io::Result<()> {
|
||||
successful_acquisition()?;
|
||||
transient_captured_child_preserves_streams_and_exit()?;
|
||||
acknowledged_transition_precedes_exact_wrapper_release()?;
|
||||
failed_acknowledged_transition_refuses_exact_wrapper_release()?;
|
||||
lifetime_pipe_descriptor_contract()?;
|
||||
|
|
@ -56,6 +57,47 @@ fn run_all() -> std::io::Result<()> {
|
|||
concurrent_acquisitions_serialize()
|
||||
}
|
||||
|
||||
fn transient_captured_child_preserves_streams_and_exit() -> std::io::Result<()> {
|
||||
let target = TargetCommand::new("/bin/sh")
|
||||
.argument("-c")
|
||||
.argument("printf transient-out; printf transient-err >&2; exit 7");
|
||||
let mut child = child_broker().run_transient_captured(&target)?;
|
||||
let mut stdout = Vec::new();
|
||||
let mut stderr = Vec::new();
|
||||
loop {
|
||||
let (out, err) = child.streams();
|
||||
let out_done = read_available(out, &mut stdout)?;
|
||||
let err_done = read_available(err, &mut stderr)?;
|
||||
if out_done && err_done && child.process().has_exited()? {
|
||||
break;
|
||||
}
|
||||
std::thread::yield_now();
|
||||
}
|
||||
let exit = child_broker().reap_status(child.into_process())?;
|
||||
if stdout != b"transient-out"
|
||||
|| stderr != b"transient-err"
|
||||
|| exit.exit_code != Some(7)
|
||||
|| exit.terminating_signal.is_some()
|
||||
{
|
||||
return Err(std::io::Error::other(
|
||||
"transient captured child changed streams or exit",
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn read_available(file: &mut File, output: &mut Vec<u8>) -> std::io::Result<bool> {
|
||||
let mut buffer = [0; 4096];
|
||||
loop {
|
||||
match file.read(&mut buffer) {
|
||||
Ok(0) => return Ok(true),
|
||||
Ok(length) => output.extend_from_slice(&buffer[..length]),
|
||||
Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => return Ok(false),
|
||||
Err(error) => return Err(error),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::disallowed_methods, clippy::disallowed_types)]
|
||||
fn failed_acknowledged_transition_refuses_exact_wrapper_release() -> std::io::Result<()> {
|
||||
let target_marker = marker("failed-transition-target");
|
||||
|
|
|
|||
Loading…
Reference in a new issue