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}; 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, command: Option, identity: ClientSessionIdentity, coordinator: Option, bootstrap_owner: Option, } 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> { 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 { 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 { 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 { 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.bootstrap_status().map_err(failure)? { b'L' => { match child.bootstrap_status().map_err(failure)? { b'R' => {} b'E' => { let result = child.reap().map_err(failure)?; return Err(failure(format!( "Host Coordinator refused image preparation with {result:?}" ))); } _ => { return Err(failure( "Host Coordinator sent an invalid image-preparation status", )); } } 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")), } } fn observe_domain( &mut self, initial_path: ObservationPath, ) -> Result> { 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) .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 startup_timeout = self .command .as_ref() .ok_or_else(|| ClientError::Backend(failure("client command was not validated")))? .startup_timeout; let mut observer = DomainObservation::new( initial_path, SystemObservationClock::start(), startup_timeout, ); 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 => {} } observer .retry() .map_err(|_| observation_deadline(observer.path()))?; } } fn attempt_waiter( &self, observer: &DomainObservation, socket: &std::path::Path, request: &BootstrapWaitRequest, ) -> Result, ClientError> { 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| { ClientError::Backend(failure(format!("bootstrap waiter request failed: {error}"))) })?; let first: BootstrapWaitOutcome = read_waiter_frame(&mut stream).map_err(|error| { 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) { 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, ) -> Result> { observer .remaining() .map_err(|_| observation_deadline(observer.path())) } fn observation_deadline(path: ObservationPath) -> ClientError { 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> { 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, 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 { let command = self .command .clone() .ok_or_else(|| failure("client command was not validated"))?; self.launch_coordinator(&command) } fn bootstrap(&mut self) -> Result> { self.observe_domain(ObservationPath::BootstrapLeader) } fn wait_until_ready(&mut self) -> Result> { self.observe_domain(ObservationPath::WaitingClient) } fn execute( &mut self, endpoint: Self::Endpoint, argv: &[Vec], stdout: &mut dyn Write, stderr: &mut dyn Write, ) -> Result> { 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)? }; self.bootstrap_owner.take(); let reap_coordinator = if let Some(coordinator) = self.coordinator.as_ref() { !matches!( self.discover().map_err(ClientError::Backend)?, Discovery::Ready(_) ) || coordinator .has_exited() .map_err(failure) .map_err(ClientError::Backend)? } else { false }; if reap_coordinator { let coordinator = self .coordinator .take() .expect("checked coordinator remains present"); 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:?}" )))); } } Ok(exit) } } pub(crate) fn open_private_domain_lock(root: &impl AsFd) -> Result { 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> { 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 { 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 { std::env::var_os(name) .filter(|value| !value.is_empty()) .map(PathBuf::from) } fn environment() -> BTreeMap { 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), 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::() .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())) } }