lazaret/crates/lazaret-client/src/coordinator_backend.rs

710 lines
27 KiB
Rust
Raw Normal View History

use std::collections::BTreeMap;
use std::ffi::OsString;
use std::fmt;
use std::fs::File;
use std::io::{self, IsTerminal, Write};
use std::os::fd::{AsFd, AsRawFd};
use std::os::unix::fs::MetadataExt;
use std::path::PathBuf;
use std::time::Duration;
use lazaret_control_client::{ControlClient, ControlClientError};
use lazaret_core::configuration::{
ConfigurationLoadRequest, ResolvedDomainConfiguration, load_trusted_configuration,
};
use lazaret_core::protocol::{
BOOTSTRAP_WAITER_SOCKET_NAME, BootstrapWaitOutcome, BootstrapWaitRequest,
PRIVATE_CLIENT_BOOTSTRAP_ROLE, read_waiter_frame, write_waiter_frame,
};
use lazaret_linux::filesystem::try_exclusive_flock;
use lazaret_linux::platform::{PlatformObservation, check_platform};
use lazaret_linux::process::{BootstrapChild, TargetCommand, child_broker};
use rustix::fs::{Mode, OFlags, openat};
use rustix::pipe::{PipeFlags, pipe_with};
2026-07-19 03:25:27 +09:00
use crate::domain_observation::{
DiscoveryAction, DiscoveryState, DomainObservation, ObservationPath, SystemObservationClock,
WaiterAction,
};
use crate::{
BootstrapAcquisition, ClientBackend, ClientCommand, ClientError, ClientSessionIdentity,
ControlEndpoint, DomainStatus, execute_interactive_over_control_face_as,
execute_over_control_face_as,
};
#[derive(Debug)]
pub struct CoordinatorClientError(String);
impl fmt::Display for CoordinatorClientError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.0)
}
}
impl std::error::Error for CoordinatorClientError {}
fn failure(error: impl fmt::Display) -> CoordinatorClientError {
CoordinatorClientError(error.to_string())
}
pub struct CoordinatorClientBackend {
configuration: Option<ResolvedDomainConfiguration>,
command: Option<ClientCommand>,
identity: ClientSessionIdentity,
coordinator: Option<BootstrapChild>,
bootstrap_owner: Option<rustix::fd::OwnedFd>,
}
impl CoordinatorClientBackend {
/// Creates one client invocation and its host-private session identity.
///
/// # Errors
/// Returns an error when operating-system randomness is unavailable.
pub fn new() -> Result<Self, ClientError<CoordinatorClientError>> {
Ok(Self {
configuration: None,
command: None,
identity: ClientSessionIdentity::generate()?,
coordinator: None,
bootstrap_owner: None,
})
}
fn configuration(&self) -> Result<&ResolvedDomainConfiguration, CoordinatorClientError> {
self.configuration
.as_ref()
.ok_or_else(|| failure("client configuration was not validated"))
}
fn discover(&self) -> Result<Discovery, CoordinatorClientError> {
let configuration = self.configuration()?;
match ControlClient::discover_confined(
configuration.runtime_domain_root(),
rustix::process::getuid().as_raw(),
) {
Ok((authority, socket_path, publication))
if publication_matches(configuration, &publication) =>
{
if !control_socket_is_live(&socket_path) {
return Ok(Discovery::Stale);
}
Ok(Discovery::Ready(Box::new(ControlEndpoint::confined(
socket_path,
publication,
authority,
))))
}
Err(ControlClientError::Io(error)) if error.kind() == io::ErrorKind::NotFound => {
Ok(Discovery::Absent)
}
Ok(_) | Err(_) => Ok(Discovery::Stale),
}
}
fn domain_lock_is_available(&self) -> Result<bool, CoordinatorClientError> {
let configuration = self.configuration()?;
let root = configuration
.persistent_domain_root()
.open_authority()
.map_err(failure)?;
let file = open_private_domain_lock(&root)?;
match try_exclusive_flock(file) {
Ok(lock) => {
drop(lock);
Ok(true)
}
Err(error) if error.kind() == io::ErrorKind::WouldBlock => Ok(false),
Err(error) => Err(failure(error)),
}
}
fn launch_coordinator(
&mut self,
command: &ClientCommand,
) -> Result<BootstrapAcquisition, CoordinatorClientError> {
let executable = std::env::current_exe().map_err(failure)?;
let coordinator = executable
.parent()
.ok_or_else(|| failure("Lazaret executable has no installation directory"))?
.join("lazaret-host-coordinator");
let (reader, writer) = pipe_with(PipeFlags::CLOEXEC).map_err(failure)?;
let mut target = TargetCommand::new(coordinator)
.argument(PRIVATE_CLIENT_BOOTSTRAP_ROLE)
.argument("--profile")
.argument(&command.profile)
.argument("--workspace")
.argument(command.workspace.as_os_str())
.with_lifetime_pipe_parent_edge();
if let Some(configuration) = &command.configuration {
target = target
.argument("--config")
.argument(configuration.as_os_str());
}
target = target
.argument("--client-id")
.argument(hex(self.identity.cli_id().as_bytes()))
.argument("--session-id")
.argument(hex(self.identity.session_id().as_bytes()))
.argument("--anchor-id")
.argument(hex(self.identity.session_anchor_id().as_bytes()))
.argument("--domain-id")
.argument(hex(self.configuration()?.domain_id().as_bytes()))
.argument("--configuration-fingerprint")
.argument(hex(self
.configuration()?
.configuration_fingerprint()
.as_bytes()));
let mut child = child_broker()
.spawn_bootstrap(&target, File::from(reader))
.map_err(failure)?;
match child.acquisition_status().map_err(failure)? {
b'L' => {
self.coordinator = Some(child);
self.bootstrap_owner = Some(writer);
Ok(BootstrapAcquisition::Leader)
}
b'W' => {
child.reap().map_err(failure)?;
Ok(BootstrapAcquisition::Waiter)
}
b'E' => {
let result = child.reap().map_err(failure)?;
Err(failure(format!(
"Host Coordinator refused bootstrap with {result:?}"
)))
}
_ => Err(failure("Host Coordinator sent an invalid bootstrap status")),
}
}
2026-07-19 03:25:27 +09:00
fn observe_domain(
&mut self,
2026-07-19 03:25:27 +09:00
initial_path: ObservationPath,
) -> Result<ControlEndpoint, ClientError<CoordinatorClientError>> {
2026-07-19 03:25:27 +09:00
let waiter = if initial_path == ObservationPath::WaitingClient {
let configuration = self.configuration().map_err(ClientError::Backend)?;
let runtime_authority = configuration
.runtime_domain_root()
.open_authority()
.map_err(failure)
2026-07-19 03:25:27 +09:00
.map_err(ClientError::Backend)?;
let socket = PathBuf::from(format!(
"/proc/self/fd/{}/{}",
runtime_authority.as_raw_fd(),
BOOTSTRAP_WAITER_SOCKET_NAME
));
let request = BootstrapWaitRequest {
protocol_version: lazaret_core::records::ProtocolVersion::V1,
domain_id: configuration.domain_id(),
configuration_fingerprint: configuration.configuration_fingerprint(),
cli_id: self.identity.cli_id(),
session_id: self.identity.session_id(),
session_anchor_id: self.identity.session_anchor_id(),
};
Some((runtime_authority, socket, request))
} else {
None
};
let mut observer = DomainObservation::new(initial_path, SystemObservationClock::start());
let mut waiter_transfer = None;
loop {
let (state, endpoint) = match self.discover().map_err(ClientError::Backend)? {
Discovery::Ready(endpoint) => (DiscoveryState::Ready, Some(*endpoint)),
Discovery::Stale => (DiscoveryState::Stale, None),
Discovery::Absent => (DiscoveryState::Absent, None),
};
match observer.discovery(state) {
DiscoveryAction::ReturnReady => {
let endpoint = endpoint.expect("ready discovery supplies an endpoint");
return Ok(match waiter_transfer {
Some(stream) => endpoint.retain_waiter_transfer(stream),
None => endpoint,
});
}
DiscoveryAction::RefuseStale => return Err(ClientError::StaleCoordinator),
DiscoveryAction::CheckCoordinator => {
let exited = self
.coordinator
.as_ref()
.ok_or_else(|| ClientError::Backend(failure("Host Coordinator is absent")))?
.has_exited()
.map_err(failure)
.map_err(ClientError::Backend)?;
if observer.coordinator_exited(exited) {
let status = self
.coordinator
.take()
.expect("checked coordinator remains present")
.reap()
.map_err(failure)
.map_err(ClientError::Backend)?;
return Err(ClientError::Backend(failure(format!(
"Host Coordinator exited before publication with {status:?}"
))));
}
}
DiscoveryAction::AttemptWaiter => {
let (_, socket, request) = waiter
.as_ref()
.expect("waiting observation retains waiter authority");
if let Some(stream) = self.attempt_waiter(&observer, socket, request)? {
waiter_transfer = Some(stream);
observer.transfer_publication();
continue;
}
}
DiscoveryAction::Retry => {}
}
2026-07-19 03:25:27 +09:00
observer
.retry()
.map_err(|_| observation_deadline(observer.path()))?;
}
}
fn attempt_waiter(
&self,
observer: &DomainObservation<SystemObservationClock>,
socket: &std::path::Path,
request: &BootstrapWaitRequest,
) -> Result<Option<std::os::unix::net::UnixStream>, ClientError<CoordinatorClientError>> {
let mut stream = match std::os::unix::net::UnixStream::connect(socket) {
Ok(stream) => stream,
Err(error)
if matches!(
error.kind(),
io::ErrorKind::NotFound | io::ErrorKind::ConnectionRefused
) =>
{
if observer.lock_available(
self.domain_lock_is_available()
.map_err(ClientError::Backend)?,
) == Some(WaiterAction::RefuseFailed)
{
return Err(ClientError::WaitingDomainFailed);
}
return Ok(None);
}
Err(error) => return Err(ClientError::Backend(failure(error))),
};
stream
.set_read_timeout(Some(observation_remaining(observer)?))
.map_err(failure)
.map_err(ClientError::Backend)?;
write_waiter_frame(&mut stream, request).map_err(|error| {
2026-07-19 03:25:27 +09:00
ClientError::Backend(failure(format!("bootstrap waiter request failed: {error}")))
})?;
let first: BootstrapWaitOutcome = read_waiter_frame(&mut stream).map_err(|error| {
2026-07-19 03:25:27 +09:00
ClientError::Backend(failure(format!(
"bootstrap waiter admission failed: {error}"
)))
})?;
let mut action = observer.waiter(first, false);
if action == WaiterAction::AwaitTerminal {
action = match read_waiter_frame(&mut stream) {
2026-07-19 03:25:27 +09:00
Ok(outcome) => observer.waiter(outcome, true),
Err(_) => WaiterAction::RefuseFailed,
};
}
match action {
WaiterAction::AwaitPublication => Ok(Some(stream)),
WaiterAction::RefuseStale => Err(ClientError::StaleCoordinator),
WaiterAction::RefuseFailed => Err(ClientError::WaitingDomainFailed),
WaiterAction::RefuseProtocol | WaiterAction::AwaitTerminal => Err(
ClientError::Protocol("bootstrap waiter sent an invalid terminal outcome"),
),
}
}
}
fn observation_remaining(
observer: &DomainObservation<SystemObservationClock>,
) -> Result<Duration, ClientError<CoordinatorClientError>> {
observer
.remaining()
.map_err(|_| observation_deadline(observer.path()))
}
fn observation_deadline(path: ObservationPath) -> ClientError<CoordinatorClientError> {
match path {
ObservationPath::BootstrapLeader => {
ClientError::Backend(failure("Host Coordinator publication deadline expired"))
}
ObservationPath::WaitingClient | ObservationPath::PublicationTransfer => {
ClientError::WaitingDomainFailed
}
}
}
impl ClientBackend for CoordinatorClientBackend {
type Endpoint = ControlEndpoint;
type Error = CoordinatorClientError;
fn validate(&mut self, command: &ClientCommand) -> Result<(), ClientError<Self::Error>> {
let kernel_release = std::fs::read_to_string("/proc/sys/kernel/osrelease")
.map_err(|error| ClientError::UnsupportedHost(error.to_string()))?;
check_platform(&PlatformObservation {
operating_system: std::env::consts::OS,
architecture: std::env::consts::ARCH,
kernel_release: kernel_release.trim(),
})
.map_err(|error| ClientError::UnsupportedHost(error.to_string()))?;
self.configuration = Some(
load_configuration(command)
.map_err(|error| ClientError::Configuration(error.to_string()))?,
);
self.command = Some(command.clone());
Ok(())
}
fn observe(&mut self) -> Result<DomainStatus<Self::Endpoint>, Self::Error> {
match self.discover()? {
Discovery::Ready(endpoint) => return Ok(DomainStatus::Ready(*endpoint)),
Discovery::Stale => return Ok(DomainStatus::Stale),
Discovery::Absent => {}
}
Ok(DomainStatus::Absent)
}
fn acquire_bootstrap(&mut self) -> Result<BootstrapAcquisition, Self::Error> {
let command = self
.command
.clone()
.ok_or_else(|| failure("client command was not validated"))?;
self.launch_coordinator(&command)
}
fn bootstrap(&mut self) -> Result<Self::Endpoint, ClientError<Self::Error>> {
2026-07-19 03:25:27 +09:00
self.observe_domain(ObservationPath::BootstrapLeader)
}
fn wait_until_ready(&mut self) -> Result<Self::Endpoint, ClientError<Self::Error>> {
2026-07-19 03:25:27 +09:00
self.observe_domain(ObservationPath::WaitingClient)
}
fn execute(
&mut self,
endpoint: Self::Endpoint,
argv: &[Vec<u8>],
stdout: &mut dyn Write,
stderr: &mut dyn Write,
) -> Result<i32, ClientError<Self::Error>> {
let exit = if std::io::stdin().is_terminal() && std::io::stdout().is_terminal() {
let mut terminal = crate::terminal::SystemTerminal::open()?;
execute_interactive_over_control_face_as(
&endpoint,
argv,
&mut terminal,
&self.identity,
)?
} else {
execute_over_control_face_as(&endpoint, argv, stdout, stderr, &self.identity)?
};
if let Some(coordinator) = self.coordinator.take() {
let status = coordinator
.reap()
.map_err(failure)
.map_err(ClientError::Backend)?;
if status.exit_code != Some(0) || status.terminating_signal.is_some() {
return Err(ClientError::Backend(failure(format!(
"Host Coordinator teardown failed with {status:?}"
))));
}
}
self.bootstrap_owner.take();
Ok(exit)
}
}
pub(crate) fn open_private_domain_lock(root: &impl AsFd) -> Result<File, CoordinatorClientError> {
let file = File::from(
openat(
root,
"domain.lock",
OFlags::RDWR | OFlags::CREATE | OFlags::CLOEXEC | OFlags::NOFOLLOW,
Mode::RUSR | Mode::WUSR,
)
.map_err(failure)?,
);
let metadata = file.metadata().map_err(failure)?;
if !metadata.file_type().is_file()
|| metadata.uid() != rustix::process::getuid().as_raw()
|| metadata.mode() & 0o7777 != 0o600
|| metadata.nlink() != 1
{
return Err(failure(
"Domain lock is not exact invoking-user mode-0600 authority",
));
}
Ok(file)
}
fn control_socket_is_live(socket_path: &std::path::Path) -> bool {
std::os::unix::net::UnixStream::connect(socket_path).is_ok()
}
fn publication_matches(
configuration: &ResolvedDomainConfiguration,
publication: &lazaret_core::records::ProtocolPublicationIdentity,
) -> bool {
publication.generation_identity.domain_id == configuration.domain_id()
&& publication.generation_identity.configuration_fingerprint
== configuration.configuration_fingerprint()
}
pub(crate) fn load_configuration(
command: &ClientCommand,
) -> Result<ResolvedDomainConfiguration, Box<dyn std::error::Error>> {
Ok(load_trusted_configuration(ConfigurationLoadRequest {
explicit_configuration_path: command.configuration.clone(),
selected_profile: command.profile.clone(),
workspace: command.workspace.clone(),
home: required_environment_path("HOME")?,
xdg_config_home: optional_environment_path("XDG_CONFIG_HOME"),
xdg_data_home: optional_environment_path("XDG_DATA_HOME"),
xdg_runtime_dir: required_environment_path("XDG_RUNTIME_DIR")?,
environment: environment(),
sandbox_layout: lazaret_sandbox_contract::default_sandbox_layout()?,
})?)
}
fn required_environment_path(name: &str) -> Result<PathBuf, CoordinatorClientError> {
std::env::var_os(name)
.filter(|value| !value.is_empty())
.map(PathBuf::from)
.ok_or_else(|| failure(format!("{name} is not set")))
}
fn optional_environment_path(name: &str) -> Option<PathBuf> {
std::env::var_os(name)
.filter(|value| !value.is_empty())
.map(PathBuf::from)
}
fn environment() -> BTreeMap<String, OsString> {
std::env::vars_os()
.filter_map(|(name, value)| name.into_string().ok().map(|name| (name, value)))
.collect()
}
fn hex(bytes: &[u8]) -> String {
use fmt::Write as _;
bytes.iter().fold(String::new(), |mut output, byte| {
write!(output, "{byte:02x}").expect("writing to a String cannot fail");
output
})
}
enum Discovery {
Absent,
Ready(Box<ControlEndpoint>),
Stale,
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use std::fs;
use std::io::Read;
use std::os::fd::AsRawFd;
use std::os::unix::ffi::OsStringExt;
use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::{UnixListener, UnixStream};
use super::*;
use lazaret_core::codec::encode;
use lazaret_core::identity::{
CertificateDigest, CoordinatorId, GenerationId, SandboxId, TrustManifestDigest,
};
use lazaret_core::records::{GenerationIdentity, ProtocolPublicationIdentity, ProtocolVersion};
fn test_artifact_root() -> std::path::PathBuf {
let root =
std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../target/t/client");
fs::create_dir_all(&root).unwrap();
root.canonicalize().unwrap()
}
#[test]
fn replaced_domain_roots_cannot_redirect_client_lock_or_control_authority() {
let persistent_tree = tempfile::Builder::new()
.prefix("lz-")
.tempdir_in(test_artifact_root())
.unwrap();
let persistent_configuration = configuration(persistent_tree.path());
let persistent_path = path(
persistent_configuration
.persistent_domain_root()
.canonical_label()
.as_bytes(),
);
fs::rename(&persistent_path, persistent_path.with_extension("old")).unwrap();
fs::create_dir(&persistent_path).unwrap();
fs::set_permissions(&persistent_path, fs::Permissions::from_mode(0o700)).unwrap();
assert!(
persistent_configuration
.persistent_domain_root()
.open_authority()
.is_err()
);
assert!(!persistent_path.join("domain.lock").exists());
let runtime_tree = tempfile::Builder::new()
.prefix("lz-")
.tempdir_in(test_artifact_root())
.unwrap();
let runtime_configuration = configuration(runtime_tree.path());
let runtime_path = path(
runtime_configuration
.runtime_domain_root()
.canonical_label()
.as_bytes(),
);
fs::rename(&runtime_path, runtime_path.with_extension("old")).unwrap();
fs::create_dir(&runtime_path).unwrap();
fs::set_permissions(&runtime_path, fs::Permissions::from_mode(0o700)).unwrap();
assert!(
ControlClient::discover_confined(
runtime_configuration.runtime_domain_root(),
rustix::process::getuid().as_raw(),
)
.is_err()
);
}
#[test]
fn confined_control_discovery_connects_through_the_retained_root() {
let tree = tempfile::Builder::new()
.prefix("lz-")
.tempdir_in(test_artifact_root())
.unwrap();
let configuration = configuration(tree.path());
let runtime_path = path(
configuration
.runtime_domain_root()
.canonical_label()
.as_bytes(),
);
let runtime = fs::File::open(&runtime_path).unwrap();
let socket_path = PathBuf::from(format!("/proc/self/fd/{}/c.sock", runtime.as_raw_fd()));
let listener = UnixListener::bind(&socket_path).unwrap();
fs::set_permissions(
runtime_path.join("c.sock"),
fs::Permissions::from_mode(0o600),
)
.unwrap();
let publication = ProtocolPublicationIdentity {
protocol_version: ProtocolVersion::V1,
generation_identity: GenerationIdentity {
agent_profile: "agent".into(),
domain_id: configuration.domain_id(),
generation_id: GenerationId::new(1),
configuration_fingerprint: configuration.configuration_fingerprint(),
coordinator_id: CoordinatorId::new([3; 32]),
sandbox_id: SandboxId::new([4; 32]),
},
tls_ca_certificate_der: vec![1],
trust_manifest_sha256: TrustManifestDigest::new([5; 32]),
coordinator_tls_certificate_sha256: CertificateDigest::new([6; 32]),
};
let mut mismatch = publication.clone();
mismatch.generation_identity.configuration_fingerprint =
lazaret_core::identity::ConfigurationFingerprint::new([7; 32]);
assert!(publication_matches(&configuration, &publication));
assert!(!publication_matches(&configuration, &mismatch));
fs::write(
runtime_path.join("protocol-identity.postcard"),
encode(&publication).unwrap(),
)
.unwrap();
fs::set_permissions(
runtime_path.join("protocol-identity.postcard"),
fs::Permissions::from_mode(0o600),
)
.unwrap();
let (authority, socket, observed) = ControlClient::discover_confined(
configuration.runtime_domain_root(),
rustix::process::getuid().as_raw(),
)
.unwrap();
let client = UnixStream::connect(socket).unwrap();
let (server, _) = listener.accept().unwrap();
assert_eq!(observed, publication);
assert!(control_socket_is_live(&socket_path));
let (probe, _) = listener.accept().unwrap();
drop((authority, client, server, probe, listener));
assert!(!control_socket_is_live(&socket_path));
}
#[test]
fn waiter_transfer_remains_live_until_control_registration_acknowledges_it() {
let (client, mut coordinator) = UnixStream::pair().unwrap();
coordinator.set_nonblocking(true).unwrap();
let endpoint = ControlEndpoint::new(
PathBuf::from("unused"),
ProtocolPublicationIdentity {
protocol_version: ProtocolVersion::V1,
generation_identity: GenerationIdentity {
agent_profile: "agent".into(),
domain_id: lazaret_core::identity::DomainId::new([1; 32]),
generation_id: GenerationId::new(1),
configuration_fingerprint:
lazaret_core::identity::ConfigurationFingerprint::new([2; 32]),
coordinator_id: CoordinatorId::new([3; 32]),
sandbox_id: SandboxId::new([4; 32]),
},
tls_ca_certificate_der: vec![1],
trust_manifest_sha256: TrustManifestDigest::new([5; 32]),
coordinator_tls_certificate_sha256: CertificateDigest::new([6; 32]),
},
)
.retain_waiter_transfer(client);
assert_eq!(
coordinator.read(&mut [0_u8; 1]).unwrap_err().kind(),
io::ErrorKind::WouldBlock
);
endpoint
.acknowledge_waiter_transfer::<std::convert::Infallible>()
.unwrap();
assert_eq!(coordinator.read(&mut [0_u8; 1]).unwrap(), 0);
}
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\"
",
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: lazaret_sandbox_contract::default_sandbox_layout().unwrap(),
})
.unwrap()
}
fn path(bytes: &[u8]) -> PathBuf {
PathBuf::from(OsString::from_vec(bytes.to_vec()))
}
}