lazaret/crates/lazaret-host-coordinator/tests/runtime-composition.rs

1267 lines
46 KiB
Rust

use std::collections::BTreeMap;
use std::fs::{self, File, OpenOptions};
use std::net::{Ipv4Addr, SocketAddrV4, TcpListener, TcpStream, ToSocketAddrs};
use std::os::linux::net::SocketAddrExt;
use std::os::unix::ffi::OsStringExt;
use std::os::unix::fs::symlink;
use std::os::unix::net::{SocketAddr, UnixListener, UnixStream};
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::network::{DnsQueryExpectation, ReachabilityExpectation, derive_ipv4_policy};
use lazaret_core::records::{
Domain, DurableDecision, Generation, HostProcessOwnershipSlot, IpAddress, ParticipantRole,
ParticipantState, ProcessOwnershipCommitAck,
};
use lazaret_host_coordinator::domain_store::{DomainStoreMutation, HostCoordinatorDomainStore};
use lazaret_host_coordinator::network::NetworkCoordinator;
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::network::{DnsProbeTarget, NetworkProbeRequest, TcpProbeTarget};
use lazaret_runtime_supervisor::sandbox::{
NETWORK_NAMESPACE_PATH, RootProcess, RunscCommand, SandboxBuildRequest, SandboxBundle,
build_sandbox_bundle, production_sandbox_layout,
};
use lazaret_runtime_supervisor::supervision::{
AttachedRunscSignal, RuntimeNetworkLaunch, 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";
const SANDBOX_NETWORK_PROBE: &str = "--sandbox-network-probe";
const NETWORK_MATRIX: &str = "--network-matrix";
const HANDOFF_MATRIX: &str = "--handoff-matrix";
const BENCHMARK_DRIVER: &str = "--runtime-composition-benchmark-command";
const BENCHMARK_GUEST: &str = "--runtime-composition-benchmark-guest";
const BENCHMARK_COMPLETE: &str = "benchmark-complete";
const BENCHMARK_ERROR: &str = "benchmark-error";
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(BENCHMARK_GUEST) {
return exit(run_benchmark_guest(&arguments[1..]));
}
if arguments.first().and_then(|argument| argument.to_str()) == Some(BENCHMARK_DRIVER) {
return exit(run_benchmark_driver(&arguments[1..]));
}
if arguments.first().and_then(|argument| argument.to_str()) == Some(HOST_COORDINATOR_DRIVER) {
return exit(run_host_coordinator_driver(&arguments[1..]));
}
if arguments.first().and_then(|argument| argument.to_str()) == Some(SANDBOX_NETWORK_PROBE) {
return exit(run_sandbox_network_probe(&arguments[1..]));
}
if arguments.first().and_then(|argument| argument.to_str()) == Some(NETWORK_MATRIX) {
return match run_network_matrix() {
Ok(()) => {
println!("PASS production network composition matrix");
ExitCode::SUCCESS
}
Err(MatrixError::Skip(reason)) => {
println!("SKIP network composition: {reason}");
ExitCode::from(SKIP)
}
Err(MatrixError::Failure(error)) => {
eprintln!("network composition failed: {error}");
ExitCode::FAILURE
}
};
}
if arguments.first().and_then(|argument| argument.to_str()) == Some(HANDOFF_MATRIX) {
return match run_handoff_matrix() {
Ok(()) => {
println!("PASS final handoff replacement refusal");
ExitCode::SUCCESS
}
Err(MatrixError::Skip(reason)) => {
println!("SKIP final handoff replacement refusal: {reason}");
ExitCode::from(SKIP)
}
Err(MatrixError::Failure(error)) => {
eprintln!("final handoff replacement refusal failed: {error}");
ExitCode::FAILURE
}
};
}
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 run_benchmark_driver(arguments: &[std::ffi::OsString]) -> std::io::Result<()> {
let separator = arguments
.iter()
.position(|argument| argument == "--")
.ok_or_else(|| std::io::Error::other("benchmark command separator is missing"))?;
if separator != 8 || arguments.len() == separator + 1 {
return Err(std::io::Error::other(
"benchmark composition arguments are malformed",
));
}
let paths = arguments[..separator]
.iter()
.map(PathBuf::from)
.collect::<Vec<_>>();
let fixture = Fixture::build_benchmark(
&paths[0],
&paths[1],
&paths[2],
&paths[3],
&paths[4],
&paths[5],
&paths[6],
&paths[7],
&arguments[separator + 1..],
)
.map_err(matrix_io)?;
let coordinator = SandboxCoordinator::new(&fixture.store);
let runtime = fixture.launch_committed().map_err(matrix_io)?;
let results = fixture
.benchmark_results
.as_ref()
.ok_or_else(|| std::io::Error::other("benchmark composition result authority is absent"))?;
let complete = results.join(BENCHMARK_COMPLETE);
let error = results.join(BENCHMARK_ERROR);
for _ in 0..24_000 {
if complete.is_file() || error.is_file() {
break;
}
thread::sleep(Duration::from_millis(5));
}
let outcome = if complete.is_file() {
Ok(())
} else if error.is_file() {
Err(std::io::Error::other(fs::read_to_string(&error)?))
} else {
Err(std::io::Error::other("benchmark command timed out"))
};
runtime
.stop(&coordinator)
.map_err(|error| std::io::Error::other(error.to_string()))?;
outcome
}
#[allow(clippy::disallowed_methods, clippy::disallowed_types)]
fn run_benchmark_guest(arguments: &[std::ffi::OsString]) -> std::io::Result<()> {
if arguments.len() < 3 {
return Err(std::io::Error::other(
"benchmark guest arguments are malformed",
));
}
let results = PathBuf::from(&arguments[0]);
let output = std::process::Command::new(&arguments[1])
.args(&arguments[2..])
.output();
match output {
Ok(output) if output.status.success() => {
fs::write(results.join(BENCHMARK_COMPLETE), b"ok")?;
}
Ok(output) => fs::write(
results.join(BENCHMARK_ERROR),
format!(
"benchmark command exited with {}: {}",
output.status,
String::from_utf8_lossy(&output.stderr)
),
)?,
Err(error) => fs::write(results.join(BENCHMARK_ERROR), error.to_string())?,
}
loop {
thread::park();
}
}
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> {
require_matrix_prerequisites()?;
lost_acknowledgment_refuses_target_work()?;
replaced_handoff_authority_refuses_target_work()?;
normal_stop_and_stop_resume()?;
attached_runsc_loss_stops_guest()?;
runtime_supervisor_loss_stops_guest()?;
host_coordinator_loss_stops_guest()?;
run_network_cases()
}
fn run_network_matrix() -> Result<(), MatrixError> {
require_matrix_prerequisites()?;
run_network_cases()
}
fn run_handoff_matrix() -> Result<(), MatrixError> {
require_matrix_prerequisites()?;
replaced_handoff_authority_refuses_target_work()?;
cross_bundle_handoff_refuses_target_work()
}
fn run_network_cases() -> Result<(), MatrixError> {
failed_probe_blocks_sandbox_start()?;
networked_sandbox_view_isolated()
}
fn require_matrix_prerequisites() -> 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()
}
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/libgcc_s.so.1"),
PathBuf::from("/lib64/libm.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 replaced_handoff_authority_refuses_target_work() -> Result<(), MatrixError> {
let fixture = Fixture::new()?;
let handoff = fixture
.bundle
.capture_handoff_authority(&fixture.configuration)
.map_err(failure)?;
let mut pending = fixture.launch_pending()?;
let staged = pending
.receive_staged(&SandboxCoordinator::new(&fixture.store))
.map_err(failure)?;
let workspace = host_path(fixture.configuration.workspace().canonical_label());
let selected = workspace.with_extension("selected");
fs::rename(&workspace, &selected)?;
fs::create_dir(&workspace)?;
let marker = workspace.join("replacement-marker");
fs::write(&marker, b"unchanged")?;
if pending.acknowledge(staged, &handoff).is_ok() {
return Err(failure("replaced handoff authority released sandbox work"));
}
if fs::read(&marker)? != b"unchanged"
|| workspace.join("activity").exists()
|| selected.join("activity").exists()
{
return Err(failure("refused handoff mutated a workspace tree"));
}
Ok(())
}
fn cross_bundle_handoff_refuses_target_work() -> Result<(), MatrixError> {
let fixture = Fixture::new()?;
let handoff = fixture
.bundle
.capture_handoff_authority(&fixture.configuration)
.map_err(failure)?;
let alternate = fixture.alternate_bundle()?;
if fixture.bundle.command().arguments() != alternate.command().arguments() {
return Err(failure("cross-bundle fixture changed runsc argv"));
}
let mut pending = fixture.launch_pending_with_command(alternate.command())?;
let staged = pending
.receive_staged(&SandboxCoordinator::new(&fixture.store))
.map_err(failure)?;
if pending.acknowledge(staged, &handoff).is_ok() || fixture.activity.exists() {
return Err(failure("cross-bundle handoff 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,
benchmark_results: Option<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)
}
#[allow(clippy::too_many_arguments)]
fn build_benchmark(
root: &Path,
workspace: &Path,
host_root: &Path,
output: &Path,
repository: &Path,
cargo_home: &Path,
rustup_home: &Path,
uv_bin: &Path,
sample_arguments: &[std::ffi::OsString],
) -> Result<Self, MatrixError> {
let configuration_directory = root.join("configuration");
let runtime = root.join("runtime");
for directory in [&configuration_directory, &runtime] {
fs::create_dir_all(directory)?;
}
let configuration_path = configuration_directory.join("lazaret.toml");
let paths = [
root,
workspace,
host_root,
output,
repository,
cargo_home,
rustup_home,
uv_bin,
];
if paths.iter().any(|path| {
path.to_str()
.is_none_or(|value| value.contains(['\n', '\r', '"']))
}) {
return Err(failure("benchmark composition path is not TOML-safe"));
}
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 = [\"/tooling\", \"/fixtures\"]\nenv = {{ PATH = \"env:BENCHMARK_PATH\", CARGO_HOME = \"env:BENCHMARK_CARGO_HOME\", RUSTUP_HOME = \"env:BENCHMARK_RUSTUP_HOME\", CARGO_BUILD_JOBS = \"env:BENCHMARK_CARGO_BUILD_JOBS\" }}\nvolumes = [\n {{ host = \"{}\", container = \"/tooling/usr\", kind = \"directory\", access = \"read-only\" }},\n {{ host = \"{}\", container = \"/fixtures/host\", kind = \"directory\", access = \"read-only\" }},\n {{ host = \"{}\", container = \"/fixtures/output\", kind = \"directory\", access = \"read-write\" }},\n {{ host = \"{}\", container = \"/fixtures/repository\", kind = \"directory\", access = \"read-only\" }},\n {{ host = \"{}\", container = \"/tooling/cargo\", kind = \"directory\", access = \"read-only\" }},\n {{ host = \"{}\", container = \"/tooling/rustup\", kind = \"directory\", access = \"read-only\" }},\n {{ host = \"{}\", container = \"/tooling/uv-bin\", kind = \"directory\", access = \"read-only\" }},\n]\n",
root.join("data").display(),
Path::new("/usr").display(),
host_root.display(),
output.display(),
repository.display(),
cargo_home.display(),
rustup_home.display(),
uv_bin.display(),
),
)?;
let environment = BTreeMap::from([
(
"BENCHMARK_PATH".into(),
"/tooling/cargo/bin:/tooling/uv-bin:/usr/bin:/bin".into(),
),
("BENCHMARK_CARGO_HOME".into(), "/tooling/cargo".into()),
("BENCHMARK_RUSTUP_HOME".into(), "/tooling/rustup".into()),
("BENCHMARK_CARGO_BUILD_JOBS".into(), "2".into()),
]);
let configuration = load_trusted_configuration(ConfigurationLoadRequest {
explicit_configuration_path: Some(configuration_path),
selected_profile: "default".into(),
workspace: workspace.to_path_buf(),
home: root.join("home"),
xdg_config_home: None,
xdg_data_home: None,
xdg_runtime_dir: runtime,
environment,
sandbox_layout: production_sandbox_layout().map_err(failure)?,
})
.map_err(failure)?;
let runtime_domain = host_path(configuration.runtime_domain_root().canonical_label());
let results = output.to_path_buf();
stage_benchmark_rootfs_and_control(&runtime_domain)?;
let mut guest_arguments = vec![
BENCHMARK_GUEST.to_owned(),
"/fixtures/output".to_owned(),
"/usr/bin/python3".to_owned(),
"/fixtures/repository/tools/ignored_workspace_benchmark.py".to_owned(),
"sample".to_owned(),
];
guest_arguments.extend(
sample_arguments
.iter()
.map(|argument| argument.to_string_lossy().into_owned()),
);
let root_process =
RootProcess::new("/bin/runtime-composition", guest_arguments).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: ContainerId::new([6; 32]),
expected_network_namespace: namespace,
root_process,
terminal: None,
})
.map_err(failure)?;
Self::finish_build(
None,
configuration,
bundle,
workspace.join("activity"),
Some(results),
)
}
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());
let container_id = ContainerId::new([6; 32]);
stage_rootfs_and_control(&runtime_domain, container_id)?;
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)?;
Self::finish_build(
tree,
configuration,
bundle,
workspace.join("activity"),
None,
)
}
fn finish_build(
tree: Option<TempDir>,
configuration: ResolvedDomainConfiguration,
bundle: SandboxBundle,
activity: PathBuf,
benchmark_results: Option<PathBuf>,
) -> Result<Self, MatrixError> {
let runtime_domain = host_path(configuration.runtime_domain_root().canonical_label());
let container_id = ContainerId::new([6; 32]);
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)?;
SandboxCoordinator::new(&store)
.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,
benchmark_results,
})
}
fn launch_pending(&self) -> Result<AwaitingAttachedRunsc, MatrixError> {
self.launch_pending_with_command(self.bundle.command())
}
fn launch_pending_with_command(
&self,
command: &RunscCommand,
) -> 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,
},
command,
)?;
AwaitingAttachedRunsc::launch(&SandboxCoordinator::new(&self.store), &launch)
.map_err(failure)
}
fn alternate_bundle(&self) -> Result<SandboxBundle, MatrixError> {
let namespace =
lazaret_linux::filesystem::namespace_identity(Path::new(NETWORK_NAMESPACE_PATH))?;
build_sandbox_bundle(SandboxBuildRequest {
configuration: &self.configuration,
container_id: ContainerId::new([6; 32]),
expected_network_namespace: namespace,
root_process: RootProcess::new(
"/bin/sh",
[
"-c".to_owned(),
"echo cross-bundle >> /workspace/activity".to_owned(),
],
)
.map_err(failure)?,
terminal: None,
})
.map_err(failure)
}
fn launch_committed(&self) -> Result<CommittedSandboxRuntime, MatrixError> {
let handoff = self
.bundle
.capture_handoff_authority(&self.configuration)
.map_err(failure)?;
let mut pending = self.launch_pending()?;
let staged = pending
.receive_staged(&SandboxCoordinator::new(&self.store))
.map_err(failure)?;
pending
.acknowledge(staged, &handoff)
.map_err(failure)?
.commit(&SandboxCoordinator::new(&self.store))
.map_err(failure)
}
}
struct NetworkFixture {
_tree: TempDir,
_host_loopback: TcpListener,
_host_abstract: UnixListener,
configuration: ResolvedDomainConfiguration,
store: HostCoordinatorDomainStore,
bundle: SandboxBundle,
policy: lazaret_core::network::CanonicalDerivedNetworkPolicy,
probe_result: PathBuf,
}
impl NetworkFixture {
fn new() -> Result<Self, MatrixError> {
let tree = tempfile::tempdir()?;
let root = tree.path();
let workspace = root.join("workspace");
let runtime = root.join("runtime");
let configuration = network_configuration(root, &workspace, &runtime)?;
let runtime_domain = host_path(configuration.runtime_domain_root().canonical_label());
let container_id = ContainerId::new([16; 32]);
stage_rootfs_and_control(&runtime_domain, container_id)?;
let rootfs = runtime_domain.join("state-anchor/merged");
fs::copy(
std::env::current_exe()?,
rootfs.join("bin/lazaret-network-probe"),
)?;
fs::create_dir_all(rootfs.join("etc"))?;
fs::write(rootfs.join("etc/resolv.conf"), b"nameserver 10.0.2.3\n")?;
let host_loopback = TcpListener::bind((Ipv4Addr::LOCALHOST, 0))?;
let host_loopback_port = host_loopback.local_addr()?.port();
let abstract_name = format!("lazaret-network-v3-{}", std::process::id()).into_bytes();
let host_abstract =
UnixListener::bind_addr(&SocketAddr::from_abstract_name(&abstract_name)?)?;
let root_process = RootProcess::new(
"/bin/lazaret-network-probe",
vec![
SANDBOX_NETWORK_PROBE.into(),
host_loopback_port.to_string(),
String::from_utf8(abstract_name.clone()).map_err(failure)?,
],
)
.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 store = network_store(&configuration, &runtime_domain, container_id)?;
let policy =
derive_ipv4_policy(&configuration, [IpAddress::V4([10, 0, 2, 3])]).map_err(failure)?;
Ok(Self {
_tree: tree,
_host_loopback: host_loopback,
_host_abstract: host_abstract,
configuration,
store,
bundle,
policy,
probe_result: workspace.join("network-probe"),
})
}
fn launch(&self) -> Result<CommittedSandboxRuntime, MatrixError> {
self.launch_with_permitted([1, 1, 1, 1], 443)
}
fn launch_with_permitted(
&self,
permitted_address: [u8; 4],
permitted_port: u16,
) -> Result<CommittedSandboxRuntime, MatrixError> {
let handoff = self
.bundle
.capture_handoff_authority(&self.configuration)
.map_err(failure)?;
let launch = RuntimeSupervisorLaunch::from_networked_runsc_command(
ProcessOwnershipContext {
domain_id: self.configuration.domain_id(),
generation_id: GenerationId::new(7),
coordinator_id: CoordinatorId::new([13; 32]),
slot: HostProcessOwnershipSlot::SandboxRuntimeSupervisor,
},
self.bundle.command(),
RuntimeNetworkLaunch {
policy: &self.policy,
configuration_fingerprint: self.configuration.configuration_fingerprint(),
network_sidecar_id: NetworkSidecarId::new([14; 32]),
},
)?;
let sandbox = SandboxCoordinator::new(&self.store);
let network = NetworkCoordinator::new(&self.store);
let mut pending = AwaitingAttachedRunsc::launch_networked(&sandbox, &network, &launch)
.map_err(failure)?;
pending.receive_helper_staged(&network).map_err(failure)?;
pending
.receive_network_readiness(
&network,
&self.policy,
|namespace| NetworkProbeRequest {
domain_id: self.configuration.domain_id(),
generation_id: GenerationId::new(7),
enforcement_namespace: namespace,
denied: vec![TcpProbeTarget {
address: IpAddress::V4([10, 1, 2, 3]),
port: 443,
}],
permitted: vec![TcpProbeTarget {
address: IpAddress::V4(permitted_address),
port: permitted_port,
}],
dns: vec![DnsProbeTarget {
resolver: IpAddress::V4([10, 0, 2, 3]),
query_name: "example.com".into(),
}],
},
&ReachabilityExpectation {
denied_destinations: vec![IpAddress::V4([10, 1, 2, 3])],
permitted_destinations: vec![IpAddress::V4(permitted_address)],
dns_queries: vec![DnsQueryExpectation {
resolver: IpAddress::V4([10, 0, 2, 3]),
query_name: "example.com".into(),
}],
},
)
.map_err(failure)?;
let staged = pending.receive_staged(&sandbox).map_err(failure)?;
let (runtime, _) = pending
.acknowledge(staged, &handoff)
.map_err(failure)?
.commit_networked(&sandbox)
.map_err(failure)?;
Ok(runtime)
}
}
fn network_configuration(
root: &Path,
workspace: &Path,
runtime: &Path,
) -> Result<ResolvedDomainConfiguration, MatrixError> {
let configuration_directory = root.join("configuration");
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/network-fixture:v1\"\npersistent_dirs = []\n",
root.join("data").display()
),
)?;
load_trusted_configuration(ConfigurationLoadRequest {
explicit_configuration_path: Some(configuration_path),
selected_profile: "default".into(),
workspace: workspace.to_path_buf(),
home: root.join("home"),
xdg_config_home: None,
xdg_data_home: None,
xdg_runtime_dir: runtime.to_path_buf(),
environment: BTreeMap::new(),
sandbox_layout: production_sandbox_layout().map_err(failure)?,
})
.map_err(failure)
}
fn network_store(
configuration: &ResolvedDomainConfiguration,
runtime_domain: &Path,
container_id: ContainerId,
) -> Result<HostCoordinatorDomainStore, MatrixError> {
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([13; 32]),
network_sidecar_id: NetworkSidecarId::new([14; 32]),
sandbox_id: SandboxId::new([15; 32]),
durable_decision: DurableDecision::None,
})])
.map_err(failure)?;
SandboxCoordinator::new(&store)
.begin_intent(
GenerationId::new(7),
container_id,
&runtime_domain.join("runsc"),
BundleDigest::new([17; 32]),
current_process_identity()?,
)
.map_err(failure)?;
Ok(store)
}
fn failed_probe_blocks_sandbox_start() -> Result<(), MatrixError> {
let fixture = NetworkFixture::new()?;
if let Ok(runtime) = fixture.launch_with_permitted([203, 0, 113, 1], 9) {
runtime
.stop(&SandboxCoordinator::new(&fixture.store))
.map_err(failure)?;
return Err(failure("failed physical probe started the sandbox"));
}
let sandbox = sandbox_participant(&fixture.store);
if sandbox.state() != ParticipantState::Intent || fixture.probe_result.exists() {
Err(failure(
"failed physical probe advanced Sandbox state or guest work",
))
} else {
Ok(())
}
}
fn networked_sandbox_view_isolated() -> Result<(), MatrixError> {
let fixture = NetworkFixture::new()?;
let runtime = fixture.launch()?;
for _ in 0..6000 {
if fixture.probe_result.is_file() {
break;
}
thread::sleep(Duration::from_millis(5));
}
let result = fs::read_to_string(&fixture.probe_result)?;
for required in [
"denied=true",
"public=true",
"dns=true",
"host_loopback=false",
"abstract_socket=false",
] {
if !result.lines().any(|line| line == required) {
return Err(failure(format!(
"guest network assertion missing: {required}"
)));
}
}
runtime
.stop(&SandboxCoordinator::new(&fixture.store))
.map_err(failure)
}
fn run_sandbox_network_probe(arguments: &[std::ffi::OsString]) -> std::io::Result<()> {
let host_port = arguments
.first()
.and_then(|value| value.to_str())
.and_then(|value| value.parse::<u16>().ok())
.filter(|port| *port != 0)
.ok_or_else(|| std::io::Error::other("sandbox probe host port is invalid"))?;
let abstract_name = arguments
.get(1)
.and_then(|value| value.to_str())
.ok_or_else(|| std::io::Error::other("sandbox probe abstract name is invalid"))?;
let timeout = Duration::from_secs(3);
let denied = TcpStream::connect_timeout(
&SocketAddrV4::new(Ipv4Addr::new(10, 1, 2, 3), 443).into(),
timeout,
)
.is_err();
let public = TcpStream::connect_timeout(
&SocketAddrV4::new(Ipv4Addr::new(1, 1, 1, 1), 443).into(),
timeout,
)
.is_ok();
let dns = ("example.com", 443)
.to_socket_addrs()
.is_ok_and(|mut addresses| addresses.next().is_some());
let host_loopback = TcpStream::connect_timeout(
&SocketAddrV4::new(Ipv4Addr::new(10, 0, 2, 2), host_port).into(),
timeout,
)
.is_ok();
let abstract_socket =
UnixStream::connect_addr(&SocketAddr::from_abstract_name(abstract_name.as_bytes())?)
.is_ok();
fs::write(
"/workspace/.network-probe",
format!(
"denied={denied}\npublic={public}\ndns={dns}\nhost_loopback={host_loopback}\nabstract_socket={abstract_socket}\n"
),
)?;
fs::rename("/workspace/.network-probe", "/workspace/network-probe")?;
loop {
thread::sleep(Duration::from_secs(1));
}
}
fn stage_rootfs_and_control(
runtime_domain: &Path,
container_id: ContainerId,
) -> 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/libgcc_s.so.1", "lib64/libgcc_s.so.1"),
("/lib64/libm.so.6", "lib64/libm.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(container_id.as_bytes()));
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 stage_benchmark_rootfs_and_control(runtime_domain: &Path) -> Result<(), MatrixError> {
let rootfs = runtime_domain.join("state-anchor/merged");
for directory in [
"bin",
"tooling/usr",
"tooling/cargo",
"tooling/rustup",
"tooling/uv-bin",
"fixtures/host",
"fixtures/output",
"fixtures/repository",
] {
fs::create_dir_all(rootfs.join(directory))?;
}
fs::copy(
std::env::current_exe()?,
rootfs.join("bin/runtime-composition"),
)?;
symlink("tooling/usr", rootfs.join("usr"))?;
symlink("usr/lib", rootfs.join("lib"))?;
symlink("usr/lib64", rootfs.join("lib64"))?;
symlink("usr/sbin", rootfs.join("sbin"))?;
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
})
}