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

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::thread;
use std::time::Duration;

use lazaret_core::configuration::{
    ConfigurationLoadRequest, ResolvedDomainConfiguration, load_trusted_configuration,
};
use lazaret_host_coordinator::client_bootstrap::PRIVATE_CLIENT_BOOTSTRAP_ROLE;
use lazaret_host_coordinator::control::{ControlError, PublishedControlFace};
use lazaret_host_coordinator::waiter::{
    BOOTSTRAP_WAITER_SOCKET_NAME, BootstrapWaitOutcome, BootstrapWaitRequest, read_frame,
    write_frame,
};
use lazaret_linux::filesystem::try_exclusive_flock;
use lazaret_linux::platform::{PlatformObservation, check_platform};
use lazaret_linux::process::{BootstrapChild, TargetCommand, child_broker};
use lazaret_runtime_supervisor::sandbox::production_sandbox_layout;
use rustix::fs::{Mode, OFlags, openat};
use rustix::pipe::{PipeFlags, pipe_with};

use crate::{
    BootstrapAcquisition, ClientBackend, ClientCommand, ClientError, ClientSessionIdentity,
    ControlEndpoint, DomainStatus, execute_interactive_over_control_face_as,
    execute_over_control_face_as,
};

#[derive(Debug)]
pub struct ProductionClientError(String);

impl fmt::Display for ProductionClientError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(&self.0)
    }
}

impl std::error::Error for ProductionClientError {}

fn failure(error: impl fmt::Display) -> ProductionClientError {
    ProductionClientError(error.to_string())
}

pub struct ProductionClientBackend {
    configuration: Option<ResolvedDomainConfiguration>,
    command: Option<ClientCommand>,
    identity: ClientSessionIdentity,
    coordinator: Option<BootstrapChild>,
    bootstrap_owner: Option<rustix::fd::OwnedFd>,
}

impl ProductionClientBackend {
    /// Creates one production client invocation and its host-private session identity.
    ///
    /// # Errors
    /// Returns an error when operating-system randomness is unavailable.
    pub fn new() -> Result<Self, ClientError<ProductionClientError>> {
        Ok(Self {
            configuration: None,
            command: None,
            identity: ClientSessionIdentity::generate()?,
            coordinator: None,
            bootstrap_owner: None,
        })
    }

    fn configuration(&self) -> Result<&ResolvedDomainConfiguration, ProductionClientError> {
        self.configuration
            .as_ref()
            .ok_or_else(|| failure("client configuration was not validated"))
    }

    fn discover(&self) -> Result<Discovery, ProductionClientError> {
        let configuration = self.configuration()?;
        match PublishedControlFace::discover_confined(
            configuration.runtime_domain_root(),
            rustix::process::getuid().as_raw(),
        ) {
            Ok((authority, socket_path, publication))
                if publication_matches(configuration, &publication) =>
            {
                Ok(Discovery::Ready(Box::new(ControlEndpoint::confined(
                    socket_path,
                    publication,
                    authority,
                ))))
            }
            Err(ControlError::Io(error)) if error.kind() == io::ErrorKind::NotFound => {
                Ok(Discovery::Absent)
            }
            Ok(_) | Err(_) => Ok(Discovery::Stale),
        }
    }

    fn domain_lock_is_available(&self) -> Result<bool, ProductionClientError> {
        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, ProductionClientError> {
        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")),
        }
    }

    fn await_publication(&mut self) -> Result<ControlEndpoint, ClientError<ProductionClientError>> {
        loop {
            match self.discover().map_err(ClientError::Backend)? {
                Discovery::Ready(endpoint) => return Ok(*endpoint),
                Discovery::Stale => return Err(ClientError::StaleCoordinator),
                Discovery::Absent => {}
            }
            if self
                .coordinator
                .as_ref()
                .ok_or_else(|| ClientError::Backend(failure("Host Coordinator is absent")))?
                .has_exited()
                .map_err(failure)
                .map_err(ClientError::Backend)?
            {
                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:?}"
                ))));
            }
            thread::sleep(Duration::from_millis(10));
        }
    }
}

impl ClientBackend for ProductionClientBackend {
    type Endpoint = ControlEndpoint;
    type Error = ProductionClientError;

    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 => {}
        }
        if self.domain_lock_is_available()? {
            Ok(DomainStatus::Absent)
        } else {
            Ok(DomainStatus::Starting)
        }
    }

    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>> {
        self.await_publication()
    }

    fn wait_until_ready(&mut self) -> Result<Self::Endpoint, ClientError<Self::Error>> {
        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(),
        };
        loop {
            match self.discover().map_err(ClientError::Backend)? {
                Discovery::Ready(endpoint) => return Ok(*endpoint),
                Discovery::Stale => return Err(ClientError::StaleCoordinator),
                Discovery::Absent => {}
            }
            match std::os::unix::net::UnixStream::connect(&socket) {
                Ok(mut stream) => {
                    write_frame(&mut stream, &request).map_err(|error| {
                        ClientError::Backend(failure(format!(
                            "bootstrap waiter request failed: {error}"
                        )))
                    })?;
                    let first: BootstrapWaitOutcome = read_frame(&mut stream).map_err(|error| {
                        ClientError::Backend(failure(format!(
                            "bootstrap waiter admission failed: {error}"
                        )))
                    })?;
                    if first != BootstrapWaitOutcome::Waiting {
                        return match first {
                            BootstrapWaitOutcome::Refused => Err(ClientError::StaleCoordinator),
                            BootstrapWaitOutcome::Failed => Err(ClientError::WaitingDomainFailed),
                            BootstrapWaitOutcome::Published => {
                                self.await_ready_publication(Some(stream))
                            }
                            BootstrapWaitOutcome::Waiting => unreachable!(),
                        };
                    }
                    return match read_frame(&mut stream) {
                        Ok(BootstrapWaitOutcome::Published) => {
                            self.await_ready_publication(Some(stream))
                        }
                        Ok(BootstrapWaitOutcome::Failed) | Err(_) => {
                            Err(ClientError::WaitingDomainFailed)
                        }
                        Ok(BootstrapWaitOutcome::Waiting | BootstrapWaitOutcome::Refused) => {
                            Err(ClientError::Protocol(
                                "bootstrap waiter sent an invalid terminal outcome",
                            ))
                        }
                    };
                }
                Err(error)
                    if matches!(
                        error.kind(),
                        io::ErrorKind::NotFound | io::ErrorKind::ConnectionRefused
                    ) => {}
                Err(error) => return Err(ClientError::Backend(failure(error))),
            }
            if self
                .domain_lock_is_available()
                .map_err(ClientError::Backend)?
            {
                return Err(ClientError::WaitingDomainFailed);
            }
            thread::sleep(Duration::from_millis(10));
        }
    }

    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)
    }
}

impl ProductionClientBackend {
    fn await_ready_publication(
        &self,
        waiter: Option<std::os::unix::net::UnixStream>,
    ) -> Result<ControlEndpoint, ClientError<ProductionClientError>> {
        loop {
            match self.discover().map_err(ClientError::Backend)? {
                Discovery::Ready(endpoint) => {
                    let endpoint = *endpoint;
                    return Ok(match waiter {
                        Some(stream) => endpoint.retain_waiter_transfer(stream),
                        None => endpoint,
                    });
                }
                Discovery::Stale => return Err(ClientError::StaleCoordinator),
                Discovery::Absent => thread::sleep(Duration::from_millis(1)),
            }
        }
    }
}

fn open_private_domain_lock(root: &impl AsFd) -> Result<File, ProductionClientError> {
    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 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()
}

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: production_sandbox_layout()?,
    })?)
}

fn required_environment_path(name: &str) -> Result<PathBuf, ProductionClientError> {
    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::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};

    #[test]
    fn replaced_domain_roots_cannot_redirect_client_lock_or_control_authority() {
        let persistent_tree = tempfile::Builder::new()
            .prefix("lz-")
            .tempdir_in("/tmp")
            .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("/tmp")
            .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!(
            PublishedControlFace::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("/tmp")
            .unwrap();
        let configuration = configuration(tree.path());
        let runtime_path = path(
            configuration
                .runtime_domain_root()
                .canonical_label()
                .as_bytes(),
        );
        let listener = UnixListener::bind(runtime_path.join("c.sock")).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) = PublishedControlFace::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);
        drop((authority, client, server));
    }

    #[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\"\n[profiles.agent.readiness]\ndenied = [{{ address = \"10.0.0.1\", port = 443 }}]\npermitted = [{{ address = \"192.0.2.10\", port = 443 }}]\ndns = [{{ resolver = \"198.51.100.53\", query = \"example.invalid\" }}]\n",
                root.join("data").display()
            ),
        )
        .unwrap();
        load_trusted_configuration(ConfigurationLoadRequest {
            explicit_configuration_path: Some(configuration_path),
            selected_profile: "agent".into(),
            workspace,
            home: root.join("home"),
            xdg_config_home: None,
            xdg_data_home: None,
            xdg_runtime_dir: runtime,
            environment: BTreeMap::new(),
            sandbox_layout: production_sandbox_layout().unwrap(),
        })
        .unwrap()
    }

    fn path(bytes: &[u8]) -> PathBuf {
        PathBuf::from(OsString::from_vec(bytes.to_vec()))
    }
}