ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Integrate host escalation with production sessions

Browse source
This commit is contained in:
ms2k 2026-07-18 04:03:39 +09:00
commit 696dae610a
7 changed files with 1175 additions and 96 deletions
Download patch file Download diff file Expand all files Collapse all files

77
crates/lazaret-host-coordinator/src/bootstrap.rs
View file

@ -11,7 +11,7 @@ use lazaret_core::records::{
Generation, GenerationIdentity, Participant, ParticipantOwnership, ParticipantRole,
ParticipantState, PreparedArtifact,
};
use lazaret_linux::process::current_process_identity;
use lazaret_linux::process::{ProcessOwnershipContext, current_process_identity};
use lazaret_runtime_supervisor::network::NetworkProbeRequest;
use lazaret_runtime_supervisor::sandbox::SandboxBundle;
use lazaret_runtime_supervisor::supervision::RuntimeSupervisorLaunch;
@ -25,6 +25,9 @@ use crate::control::{
TrustAnchor,
};
use crate::domain_store::{DomainStoreMutation, HostCoordinatorDomainStore};
use crate::escalation::{
ProductionEscalation, ProductionEscalationHandle, ProductionEscalationInputs,
};
use crate::git_routing::GitRoutingCoordinator;
use crate::lifecycle::{
HostCoordinatorLifecycle, LifecycleAction, LifecycleAdapter, SessionAdmissionIdentity,
@ -94,6 +97,8 @@ pub struct ProductionLifecycleAdapter<'a> {
trust_anchor: Option<TrustAnchor>,
staged_control: Option<StagedControlFace>,
published_control: Option<Arc<PublishedControlFace>>,
escalation: Option<ProductionEscalation<'a>>,
escalation_handle: Option<ProductionEscalationHandle>,
awaiting_runtime: Option<AwaitingAttachedRunsc>,
runtime: Option<CommittedSandboxRuntime>,
state_anchor_mount: Option<MountIdentity>,
@ -191,6 +196,8 @@ impl<'a> ProductionLifecycleAdapter<'a> {
trust_anchor: None,
staged_control: None,
published_control: None,
escalation: None,
escalation_handle: None,
awaiting_runtime: None,
runtime: None,
state_anchor_mount: None,
@ -207,6 +214,12 @@ impl<'a> ProductionLifecycleAdapter<'a> {
self.published_control.as_ref().map(Arc::clone)
}
pub(crate) fn take_escalation(
&mut self,
) -> Option<(ProductionEscalation<'a>, ProductionEscalationHandle)> {
Some((self.escalation.take()?, self.escalation_handle.take()?))
}
fn require_generation(&self, generation: &Generation) -> Result<(), ProductionLifecycleError> {
if generation.domain_id == self.plan.generation_identity.domain_id
&& generation.generation_id == self.plan.generation_identity.generation_id
@ -382,10 +395,48 @@ impl<'a> ProductionLifecycleAdapter<'a> {
}
fn stage_publication(&mut self) -> Result<(), ProductionLifecycleError> {
let anchor = self
let mut anchor = self
.trust_anchor
.take()
.ok_or_else(|| ProductionLifecycleError("Trust Anchor is absent".into()))?;
if self
.plan
.configuration
.fingerprint_record()
.escalation_enabled
{
let publication = anchor.publication_identity();
let coordinator_key = anchor.coordinator_protocol_key();
let credentials = anchor.take_sandbox_credential_bundle().map_err(failure)?;
let (credential_files, inbox, outbox) = self
.plan
.runtime_staging
.escalation_mailboxes(self.plan.invoking_uid)
.map_err(failure)?;
let ownership = ProcessOwnershipContext {
domain_id: self.plan.generation_identity.domain_id,
generation_id: self.plan.generation_identity.generation_id,
coordinator_id: self.plan.generation_identity.coordinator_id,
slot: lazaret_core::records::HostProcessOwnershipSlot::EscalationHostProcess(
lazaret_core::identity::RequestId::new([0; 32]),
),
};
let (escalation, handle) = ProductionEscalation::new(ProductionEscalationInputs {
configuration: self.plan.configuration,
publication,
credentials,
credential_files,
inbox,
outbox,
prompt_transport: Arc::clone(&self.plan.prompt_transport),
coordinator_key,
ownership,
invoking_uid: self.plan.invoking_uid,
})
.map_err(failure)?;
self.escalation = Some(escalation);
self.escalation_handle = Some(handle);
}
let generation_id = self.plan.generation_identity.generation_id;
self.plan
.store
@ -691,6 +742,15 @@ pub(crate) fn serve_production_control_connection(
lifecycle: &Mutex<ProductionHostCoordinatorLifecycle<'_, '_>>,
face: &PublishedControlFace,
connection: AcceptedControlConnection,
) -> Result<(), ControlError> {
serve_production_control_connection_with_escalation(lifecycle, face, connection, None)
}
pub(crate) fn serve_production_control_connection_with_escalation(
lifecycle: &Mutex<ProductionHostCoordinatorLifecycle<'_, '_>>,
face: &PublishedControlFace,
connection: AcceptedControlConnection,
escalation: Option<&ProductionEscalationHandle>,
) -> Result<(), ControlError> {
face.serve_connection_coordinated(
connection,
@ -701,7 +761,20 @@ pub(crate) fn serve_production_control_connection(
.admit_session(admission)
.map_err(|error| ControlError::Lifecycle(error.to_string()))
},
|binding, registration| {
if let Some(escalation) = escalation {
escalation
.register(binding, registration)
.map_err(|error| ControlError::Lifecycle(error.to_string()))?;
}
Ok(())
},
|session_id| {
if let Some(escalation) = escalation {
escalation
.disconnect(session_id)
.map_err(|error| ControlError::Lifecycle(error.to_string()))?;
}
let mut lifecycle = lifecycle.lock().map_err(|_| ControlError::LockPoisoned)?;
lifecycle
.disconnect_session(session_id)

77
crates/lazaret-host-coordinator/src/client_bootstrap.rs
View file

@ -20,9 +20,12 @@ use lazaret_linux::filesystem::try_exclusive_flock;
use lazaret_runtime_supervisor::sandbox::production_sandbox_layout;
use rustix::fs::{Mode, OFlags, openat};
use crate::bootstrap::{ProductionHostCoordinatorLifecycle, serve_production_control_connection};
use crate::bootstrap::{
ProductionHostCoordinatorLifecycle, serve_production_control_connection_with_escalation,
};
use crate::control::ControlError;
use crate::domain_store::HostCoordinatorDomainStore;
use crate::escalation::{ProductionEscalation, ProductionEscalationHandle};
use crate::production::{BootstrapOwnerGuard, ProductionColdStart, ProductionColdStartError};
use crate::production_selector::{InstalledProductPaths, select_production_cold_start};
use crate::waiter::BootstrapWaiterFace;
@ -163,15 +166,16 @@ fn serve_published(
bootstrap_owner: &BootstrapOwnerGuard,
bootstrap_owner_session_id: SessionId,
) -> Result<(), ProductionColdStartError> {
lifecycle
.lock()
.map_err(|_| ProductionColdStartError::service("production lifecycle lock is poisoned"))?
.adapter()
.published_control()
.ok_or_else(|| ProductionColdStartError::service("control face is absent"))?
.set_nonblocking(true)
.map_err(ProductionColdStartError::service)?;
let escalation = take_escalation(lifecycle)?;
std::thread::scope(|scope| {
let escalation_enabled = escalation.is_some();
let (escalation, escalation_handle) = escalation.unzip();
let (escalation_completed, escalation_completions) = std::sync::mpsc::sync_channel(1);
if let Some(escalation) = escalation {
scope.spawn(move || {
let _ = escalation_completed.send(escalation.run());
});
}
let (completed, completions) =
std::sync::mpsc::channel::<(u64, Result<(), ControlError>)>();
let mut active: BTreeMap<u64, std::os::unix::net::UnixStream> = BTreeMap::new();
@ -181,6 +185,7 @@ fn serve_published(
active.remove(&connection_id);
connection_completion(result, false)?;
}
require_escalation_running(&escalation_completions, escalation_enabled)?;
let face = {
let mut lifecycle = lifecycle.lock().map_err(|_| {
ProductionColdStartError::service("production lifecycle lock is poisoned")
@ -193,6 +198,13 @@ fn serve_published(
while !active.is_empty() {
wait_for_connection_completion(&completions, &mut active, true)?;
}
if let Some(escalation) = &escalation_handle {
escalation.stop();
escalation_completions
.recv()
.map_err(ProductionColdStartError::service)?
.map_err(ProductionColdStartError::service)?;
}
return Ok(());
};
lifecycle
@ -224,9 +236,14 @@ fn serve_published(
})?;
active.insert(connection_id, cancellation);
let completed = completed.clone();
let escalation = escalation_handle.clone();
scope.spawn(move || {
let result =
serve_production_control_connection(lifecycle, &face, connection);
let result = serve_production_control_connection_with_escalation(
lifecycle,
&face,
connection,
escalation.as_ref(),
);
let _ = completed.send((connection_id, result));
});
}
@ -241,6 +258,44 @@ fn serve_published(
})
}
fn take_escalation<'resources>(
lifecycle: &Mutex<ProductionHostCoordinatorLifecycle<'_, 'resources>>,
) -> Result<
Option<(ProductionEscalation<'resources>, ProductionEscalationHandle)>,
ProductionColdStartError,
> {
let mut lifecycle = lifecycle
.lock()
.map_err(|_| ProductionColdStartError::service("production lifecycle lock is poisoned"))?;
lifecycle
.adapter()
.published_control()
.ok_or_else(|| ProductionColdStartError::service("control face is absent"))?
.set_nonblocking(true)
.map_err(ProductionColdStartError::service)?;
Ok(lifecycle.adapter_mut().take_escalation())
}
fn require_escalation_running(
completions: &std::sync::mpsc::Receiver<Result<(), crate::escalation::EscalationError>>,
enabled: bool,
) -> Result<(), ProductionColdStartError> {
match completions.try_recv() {
Ok(result) => {
result.map_err(ProductionColdStartError::service)?;
Err(ProductionColdStartError::service(
"production escalation service stopped before teardown",
))
}
Err(std::sync::mpsc::TryRecvError::Disconnected) if enabled => Err(
ProductionColdStartError::service("production escalation service was abandoned"),
),
Err(std::sync::mpsc::TryRecvError::Empty | std::sync::mpsc::TryRecvError::Disconnected) => {
Ok(())
}
}
}
fn wait_for_connection_completion(
completions: &std::sync::mpsc::Receiver<(u64, Result<(), ControlError>)>,
active: &mut BTreeMap<u64, std::os::unix::net::UnixStream>,

73
crates/lazaret-host-coordinator/src/control.rs
View file

@ -988,7 +988,14 @@ impl NonInteractiveSessionExecutor for RunscNonInteractiveExecutor {
for argument in &self.global_arguments {
target = target.argument(argument.clone());
}
target = target.argument("exec").argument(self.container.clone());
target = target
.argument("exec")
.argument("-env")
.argument(format!(
"LAZARET_SESSION_ID={}",
hex_lower(request.session_id.as_bytes())
))
.argument(self.container.clone());
for argument in &request.argv {
target = target.argument(OsString::from_vec(argument.clone()));
}
@ -1530,9 +1537,23 @@ impl PublishedControlFace {
where
Admit: FnMut(ControlSessionAdmission) -> Result<(), ControlError>,
Disconnect: FnMut(SessionId) -> Result<(), ControlError>,
{
self.serve_next_coordinated_with_registration(admit, |_, _| Ok(()), disconnect)
}
pub(crate) fn serve_next_coordinated_with_registration<Admit, Registered, Disconnect>(
&self,
admit: Admit,
registered: Registered,
disconnect: Disconnect,
) -> Result<(), ControlError>
where
Admit: FnMut(ControlSessionAdmission) -> Result<(), ControlError>,
Registered: FnMut(SessionBinding, SessionRegistrationResponse) -> Result<(), ControlError>,
Disconnect: FnMut(SessionId) -> Result<(), ControlError>,
{
let connection = self.accept_connection()?;
self.serve_connection_coordinated(connection, admit, disconnect)
self.serve_connection_coordinated(connection, admit, registered, disconnect)
}
pub(crate) fn accept_connection(&self) -> Result<AcceptedControlConnection, ControlError> {
@ -1540,14 +1561,16 @@ impl PublishedControlFace {
Ok(AcceptedControlConnection { socket })
}
pub(crate) fn serve_connection_coordinated<Admit, Disconnect>(
pub(crate) fn serve_connection_coordinated<Admit, Registered, Disconnect>(
&self,
connection: AcceptedControlConnection,
mut admit: Admit,
mut registered: Registered,
mut disconnect: Disconnect,
) -> Result<(), ControlError>
where
Admit: FnMut(ControlSessionAdmission) -> Result<(), ControlError>,
Registered: FnMut(SessionBinding, SessionRegistrationResponse) -> Result<(), ControlError>,
Disconnect: FnMut(SessionId) -> Result<(), ControlError>,
{
let socket = connection.socket;
@ -1571,9 +1594,13 @@ impl PublishedControlFace {
let mut stream = rustls::StreamOwned::new(connection, socket);
match peer_certificate {
None => self.serve_enrollment(&mut stream),
Some(certificates) => {
self.serve_registration(&mut stream, &certificates, &mut admit, &mut disconnect)
}
Some(certificates) => self.serve_registration(
&mut stream,
&certificates,
&mut admit,
&mut registered,
&mut disconnect,
),
}
}
@ -1596,15 +1623,17 @@ impl PublishedControlFace {
)
}
fn serve_registration<Admit, Disconnect>(
fn serve_registration<Admit, Registered, Disconnect>(
&self,
stream: &mut rustls::StreamOwned<ServerConnection, std::os::unix::net::UnixStream>,
certificates: &[Vec<u8>],
admit: &mut Admit,
registered: &mut Registered,
disconnect: &mut Disconnect,
) -> Result<(), ControlError>
where
Admit: FnMut(ControlSessionAdmission) -> Result<(), ControlError>,
Registered: FnMut(SessionBinding, SessionRegistrationResponse) -> Result<(), ControlError>,
Disconnect: FnMut(SessionId) -> Result<(), ControlError>,
{
let [leaf] = certificates else {
@ -1622,18 +1651,18 @@ impl PublishedControlFace {
&self.anchor.coordinator_protocol_key,
)?;
let session_id = request.session_binding.session_id;
let binding = request.session_binding.clone();
admit(ControlSessionAdmission {
request,
credential,
})?;
let result = (|| {
write_frame(
stream,
&ControlResponse::Registered(SessionRegistrationResponse {
protocol_version: ProtocolVersion::V1,
session_capability: capability,
}),
)?;
let registration = SessionRegistrationResponse {
protocol_version: ProtocolVersion::V1,
session_capability: capability,
};
registered(binding, registration.clone())?;
write_frame(stream, &ControlResponse::Registered(registration))?;
let request: ControlRequest = match read_frame(stream, MAX_CONTROL_BYTES) {
Ok(request) => request,
Err(ControlError::Io(error))
@ -2156,10 +2185,18 @@ mod tests {
runner.observed.lock().unwrap().as_ref().unwrap(),
&(
OsString::from("runsc"),
["--root=/runtime", "exec", "container-id", "printf", "hello"]
.into_iter()
.map(OsString::from)
.collect()
vec![
OsString::from("--root=/runtime"),
OsString::from("exec"),
OsString::from("-env"),
OsString::from(format!(
"LAZARET_SESSION_ID={}",
hex_lower(session_id.as_bytes())
)),
OsString::from("container-id"),
OsString::from("printf"),
OsString::from("hello"),
]
)
);
}

594
crates/lazaret-host-coordinator/src/escalation.rs
View file

@ -6,20 +6,23 @@ use std::io::{self, Read};
use std::os::fd::{AsFd, OwnedFd};
use std::os::unix::ffi::{OsStrExt, OsStringExt};
use std::rc::Rc;
use std::sync::{Arc, mpsc};
use std::time::{Duration, Instant};
use ed25519_dalek::SigningKey;
use lazaret_core::configuration::ResolvedDomainConfiguration;
use lazaret_core::identity::{RequestDigest, RequestId, SessionId, StreamDigest};
use lazaret_core::protocol::{
MAX_RESULT_CHUNK_BYTES, ProtocolCodecError, ProtocolValidationError, RequestReplayGuard,
ResultSenderWindow, ResultWindowError, VerifiedSession, encode_signed, result_chunk_file_name,
result_file_name,
MAX_RESULT_CHUNK_BYTES, MAX_SIGNED_RECORD_BYTES, ProtocolCodecError, ProtocolValidationError,
RequestReplayGuard, ResultSenderWindow, ResultWindowError, VerifiedSession,
acknowledgment_file_name, cancellation_file_name, encode_signed, is_ready_record,
mailbox_session_name, result_chunk_file_name, result_file_name,
};
use lazaret_core::records::{
DeveloperDecisionKind, EscalationOutcome, GenerationIdentity, HostProcessOwnershipSlot,
MailboxRequest, MailboxResult, MailboxResultChunk, ProcessOwnershipCommitAck, ProtocolVersion,
ResultStream, StagedProcessOwnership,
MailboxRequest, MailboxResult, MailboxResultChunk, ProcessOwnershipCommitAck,
ProtocolPublicationIdentity, ProtocolVersion, ResultStream, SandboxCredentialBundle,
SessionBinding, SessionRegistrationResponse, StagedProcessOwnership,
};
use lazaret_linux::filesystem::{MailboxEntryIdentity, RegularFileMailbox};
use lazaret_linux::process::{
@ -29,6 +32,8 @@ use rustix::fs::{FileType, Mode, OFlags, ResolveFlags, fchmod, fstat, mkdirat, o
use rustix::process::Signal;
use sha2::{Digest, Sha256};
use crate::terminal::{SessionPromptTransport, TerminalError};
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ApprovalSource {
AutoRule,
@ -192,6 +197,7 @@ pub enum EscalationError {
Protocol(ProtocolValidationError),
ProtocolCodec(ProtocolCodecError),
Window(ResultWindowError),
Terminal(TerminalError),
DuplicateSessionRequest,
UnknownRequest,
InvalidPhase,
@ -205,6 +211,7 @@ impl fmt::Display for EscalationError {
Self::Protocol(error) => error.fmt(formatter),
Self::ProtocolCodec(error) => error.fmt(formatter),
Self::Window(error) => error.fmt(formatter),
Self::Terminal(error) => error.fmt(formatter),
Self::DuplicateSessionRequest => {
formatter.write_str("session already has an active escalation request")
}
@ -224,6 +231,7 @@ impl std::error::Error for EscalationError {
Self::Protocol(error) => Some(error),
Self::ProtocolCodec(error) => Some(error),
Self::Window(error) => Some(error),
Self::Terminal(error) => Some(error),
_ => None,
}
}
@ -253,6 +261,27 @@ impl From<ResultWindowError> for EscalationError {
}
}
impl From<TerminalError> for EscalationError {
fn from(error: TerminalError) -> Self {
Self::Terminal(error)
}
}
impl EscalationError {
fn is_record_refusal(&self) -> bool {
matches!(
self,
Self::Protocol(_)
| Self::ProtocolCodec(_)
| Self::Window(_)
| Self::DuplicateSessionRequest
| Self::UnknownRequest
| Self::InvalidPhase
| Self::InvalidApproval
)
}
}
struct RequestAuthority {
session: VerifiedSession,
request: MailboxRequest,
@ -263,6 +292,8 @@ struct RequestAuthority {
struct PublishedChunk {
name: OsString,
identity: MailboxEntryIdentity,
stream: ResultStream,
bytes: Vec<u8>,
}
struct Execution {
@ -270,13 +301,20 @@ struct Execution {
process: CapturedProcess,
window: ResultSenderWindow,
chunks: BTreeMap<u64, PublishedChunk>,
stdout: Sha256,
stderr: Sha256,
acknowledged_stdout: Sha256,
acknowledged_stderr: Sha256,
stdout_eof: bool,
stderr_eof: bool,
acknowledgment_deadline: Option<Instant>,
}
struct TerminatedExecution {
authority: RequestAuthority,
chunk_count: u64,
stdout_sha256: StreamDigest,
stderr_sha256: StreamDigest,
}
enum ActiveRequest {
AwaitingApproval(Box<RequestAuthority>),
Executing(Box<Execution>),
@ -471,9 +509,24 @@ impl EscalationCoordinator {
if decision.approved_command_argv.is_empty() {
return Err(EscalationError::InvalidApproval);
}
let executable = resolver.resolve(&decision.approved_command_argv)?;
validate_executable(executable.as_fd(), self.invoking_uid)?;
Some(executable)
if let Ok(executable) =
resolver
.resolve(&decision.approved_command_argv)
.and_then(|executable| {
validate_executable(executable.as_fd(), self.invoking_uid)?;
Ok(executable)
})
{
Some(executable)
} else {
let Some(ActiveRequest::AwaitingApproval(authority)) =
self.active.remove(&request_id)
else {
return Err(EscalationError::InvalidPhase);
};
self.finish(&authority, EscalationOutcome::Failed, None)?;
return Ok(EscalationPhase::Delivered(EscalationOutcome::Failed));
}
}
};
let Some(ActiveRequest::AwaitingApproval(authority)) = self.active.remove(&request_id)
@ -549,8 +602,8 @@ impl EscalationCoordinator {
process,
window: ResultSenderWindow::default(),
chunks: BTreeMap::new(),
stdout: Sha256::new(),
stderr: Sha256::new(),
acknowledged_stdout: Sha256::new(),
acknowledged_stderr: Sha256::new(),
stdout_eof: false,
stderr_eof: false,
acknowledgment_deadline: None,
@ -581,8 +634,8 @@ impl EscalationCoordinator {
};
let progress = self.read_available(&mut execution, now);
if progress.is_err() {
let authority = self.terminate(*execution)?;
self.finish(&authority, EscalationOutcome::Failed, None)?;
let terminated = self.terminate(*execution)?;
self.finish_terminated(&terminated, EscalationOutcome::Failed)?;
return Ok(EscalationPhase::Delivered(EscalationOutcome::Failed));
}
if execution.process.process().has_exited()?
@ -592,9 +645,9 @@ impl EscalationCoordinator {
{
let execution = *execution;
let authority = execution.authority;
let stdout = StreamDigest::new(execution.stdout.finalize().into());
let stderr = StreamDigest::new(execution.stderr.finalize().into());
let chunk_count = execution.window.sent_count();
let stdout = StreamDigest::new(execution.acknowledged_stdout.finalize().into());
let stderr = StreamDigest::new(execution.acknowledged_stderr.finalize().into());
let chunk_count = execution.window.acknowledged_count();
let status = child_broker().reap_status(execution.process.into_process())?;
self.finish_with_streams(
&authority,
@ -656,16 +709,18 @@ impl EscalationCoordinator {
stream,
bytes,
};
match stream {
ResultStream::Stdout => execution.stdout.update(&chunk.bytes),
ResultStream::Stderr => execution.stderr.update(&chunk.bytes),
}
let name = result_chunk_file_name(chunk.request_id, sequence);
let encoded = encode_signed(&chunk, &self.coordinator_key)?;
let identity = self.outbox.publish(&name, &encoded)?;
execution
.chunks
.insert(sequence, PublishedChunk { name, identity });
execution.chunks.insert(
sequence,
PublishedChunk {
name,
identity,
stream,
bytes: chunk.bytes,
},
);
if execution.acknowledgment_deadline.is_none() {
execution.acknowledgment_deadline =
Some(now + self.acknowledgment_timeout);
@ -712,7 +767,12 @@ impl EscalationCoordinator {
for sequence in acknowledged {
if let Some(chunk) = execution.chunks.get(&sequence) {
self.outbox.remove_if_matches(&chunk.name, chunk.identity)?;
execution.chunks.remove(&sequence);
}
if let Some(chunk) = execution.chunks.remove(&sequence) {
match chunk.stream {
ResultStream::Stdout => execution.acknowledged_stdout.update(&chunk.bytes),
ResultStream::Stderr => execution.acknowledged_stderr.update(&chunk.bytes),
}
}
}
execution.acknowledgment_deadline = if execution.window.outstanding() == 0 {
@ -752,11 +812,15 @@ impl EscalationCoordinator {
.active
.remove(&request_id)
.ok_or(EscalationError::UnknownRequest)?;
let authority = match active {
ActiveRequest::AwaitingApproval(authority) => *authority,
ActiveRequest::Executing(execution) => self.terminate(*execution)?,
};
self.finish(&authority, EscalationOutcome::Cancelled, None)?;
match active {
ActiveRequest::AwaitingApproval(authority) => {
self.finish(&authority, EscalationOutcome::Cancelled, None)?;
}
ActiveRequest::Executing(execution) => {
let terminated = self.terminate(*execution)?;
self.finish_terminated(&terminated, EscalationOutcome::Cancelled)?;
}
}
Ok(EscalationPhase::Delivered(EscalationOutcome::Cancelled))
}
@ -789,11 +853,15 @@ impl EscalationCoordinator {
.active
.remove(request_id)
.ok_or(EscalationError::UnknownRequest)?;
let authority = match active {
ActiveRequest::AwaitingApproval(authority) => *authority,
ActiveRequest::Executing(execution) => self.terminate(*execution)?,
};
self.finish(&authority, EscalationOutcome::TimedOut, None)?;
match active {
ActiveRequest::AwaitingApproval(authority) => {
self.finish(&authority, EscalationOutcome::TimedOut, None)?;
}
ActiveRequest::Executing(execution) => {
let terminated = self.terminate(*execution)?;
self.finish_terminated(&terminated, EscalationOutcome::TimedOut)?;
}
}
}
Ok(expired)
}
@ -829,14 +897,37 @@ impl EscalationCoordinator {
Ok(())
}
fn terminate(&mut self, execution: Execution) -> Result<RequestAuthority, EscalationError> {
fn terminate(&mut self, execution: Execution) -> Result<TerminatedExecution, EscalationError> {
execution.process.process().send_signal(Signal::KILL)?;
let authority = execution.authority;
let chunk_count = execution.window.acknowledged_count();
let stdout_sha256 = StreamDigest::new(execution.acknowledged_stdout.finalize().into());
let stderr_sha256 = StreamDigest::new(execution.acknowledged_stderr.finalize().into());
child_broker().reap(execution.process.into_process())?;
for chunk in execution.chunks.into_values() {
self.outbox.remove_if_matches(&chunk.name, chunk.identity)?;
}
Ok(authority)
Ok(TerminatedExecution {
authority,
chunk_count,
stdout_sha256,
stderr_sha256,
})
}
fn finish_terminated(
&mut self,
execution: &TerminatedExecution,
outcome: EscalationOutcome,
) -> Result<(), EscalationError> {
self.finish_with_streams(
&execution.authority,
outcome,
None,
execution.chunk_count,
execution.stdout_sha256,
execution.stderr_sha256,
)
}
fn finish(
@ -895,6 +986,435 @@ impl EscalationCoordinator {
}
}
struct SessionAuthorityFiles {
binding_name: OsString,
binding_identity: MailboxEntryIdentity,
registration_name: OsString,
registration_identity: MailboxEntryIdentity,
}
struct ProductionSessionAuthority {
session: VerifiedSession,
files: SessionAuthorityFiles,
}
struct RegistrationCommand {
binding: SessionBinding,
registration: SessionRegistrationResponse,
completed: mpsc::SyncSender<Result<(), String>>,
}
enum ProductionEscalationCommand {
Register(Box<RegistrationCommand>),
Disconnect {
session_id: SessionId,
completed: mpsc::SyncSender<Result<(), String>>,
},
Stop,
}
#[derive(Clone)]
pub(crate) struct ProductionEscalationHandle {
commands: mpsc::Sender<ProductionEscalationCommand>,
}
impl ProductionEscalationHandle {
pub(crate) fn register(
&self,
binding: SessionBinding,
registration: SessionRegistrationResponse,
) -> Result<(), EscalationError> {
let (completed, result) = mpsc::sync_channel(1);
self.commands
.send(ProductionEscalationCommand::Register(Box::new(
RegistrationCommand {
binding,
registration,
completed,
},
)))
.map_err(|_| io::Error::other("production escalation service stopped"))?;
result
.recv()
.map_err(|_| io::Error::other("production escalation registration was abandoned"))?
.map_err(|error| io::Error::other(error).into())
}
pub(crate) fn disconnect(&self, session_id: SessionId) -> Result<(), EscalationError> {
let (completed, result) = mpsc::sync_channel(1);
self.commands
.send(ProductionEscalationCommand::Disconnect {
session_id,
completed,
})
.map_err(|_| io::Error::other("production escalation service stopped"))?;
result
.recv()
.map_err(|_| io::Error::other("production escalation disconnect was abandoned"))?
.map_err(|error| io::Error::other(error).into())
}
pub(crate) fn stop(&self) {
let _ = self.commands.send(ProductionEscalationCommand::Stop);
}
}
pub(crate) struct ProductionEscalation<'a> {
configuration: &'a ResolvedDomainConfiguration,
publication: ProtocolPublicationIdentity,
credentials: SandboxCredentialBundle,
credential_files: RegularFileMailbox,
inbox: RegularFileMailbox,
prompt_transport: Arc<SessionPromptTransport>,
coordinator: EscalationCoordinator,
ownership: LocalExecutionOwnership,
sessions: BTreeMap<SessionId, ProductionSessionAuthority>,
active: BTreeMap<RequestId, SessionId>,
commands: mpsc::Receiver<ProductionEscalationCommand>,
}
pub(crate) struct ProductionEscalationInputs<'a> {
pub configuration: &'a ResolvedDomainConfiguration,
pub publication: ProtocolPublicationIdentity,
pub credentials: SandboxCredentialBundle,
pub credential_files: RegularFileMailbox,
pub inbox: RegularFileMailbox,
pub outbox: RegularFileMailbox,
pub prompt_transport: Arc<SessionPromptTransport>,
pub coordinator_key: SigningKey,
pub ownership: ProcessOwnershipContext,
pub invoking_uid: u32,
}
impl<'a> ProductionEscalation<'a> {
pub(crate) fn new(
inputs: ProductionEscalationInputs<'a>,
) -> Result<(Self, ProductionEscalationHandle), EscalationError> {
inputs.credential_files.publish(
OsStr::new("publication.postcard"),
&lazaret_core::codec::encode(&inputs.publication)
.map_err(|error| io::Error::other(error.to_string()))?,
)?;
inputs.credential_files.publish(
OsStr::new("sandbox-credentials.postcard"),
&lazaret_core::codec::encode(&inputs.credentials)
.map_err(|error| io::Error::other(error.to_string()))?,
)?;
let (commands, receiver) = mpsc::channel();
Ok((
Self {
configuration: inputs.configuration,
publication: inputs.publication,
credentials: inputs.credentials,
credential_files: inputs.credential_files,
inbox: inputs.inbox,
prompt_transport: inputs.prompt_transport,
coordinator: EscalationCoordinator::new(
inputs.coordinator_key,
inputs.outbox,
inputs.ownership,
inputs.invoking_uid,
Duration::from_secs(30),
Duration::from_secs(2),
),
ownership: LocalExecutionOwnership::default(),
sessions: BTreeMap::new(),
active: BTreeMap::new(),
commands: receiver,
},
ProductionEscalationHandle { commands },
))
}
pub(crate) fn run(mut self) -> Result<(), EscalationError> {
loop {
while let Ok(command) = self.commands.try_recv() {
if self.apply_command(command)? {
return Ok(());
}
}
self.process_requests()?;
self.process_active()?;
std::thread::sleep(Duration::from_millis(5));
}
}
fn apply_command(
&mut self,
command: ProductionEscalationCommand,
) -> Result<bool, EscalationError> {
match command {
ProductionEscalationCommand::Register(command) => {
let RegistrationCommand {
binding,
registration,
completed,
} = *command;
let result = self
.register(&binding, &registration)
.map_err(|error| error.to_string());
let failed = result.is_err();
let _ = completed.send(result);
if failed {
return Err(io::Error::other("session escalation registration failed").into());
}
Ok(false)
}
ProductionEscalationCommand::Disconnect {
session_id,
completed,
} => {
let result = self
.disconnect(session_id)
.map_err(|error| error.to_string());
let failed = result.is_err();
let _ = completed.send(result);
if failed {
return Err(io::Error::other("session escalation cleanup failed").into());
}
Ok(false)
}
ProductionEscalationCommand::Stop => {
let sessions = self.sessions.keys().copied().collect::<Vec<_>>();
for session_id in sessions {
self.disconnect(session_id)?;
}
Ok(true)
}
}
}
fn register(
&mut self,
binding: &SessionBinding,
registration: &SessionRegistrationResponse,
) -> Result<(), EscalationError> {
let session_id = binding.session_id;
if self.sessions.contains_key(&session_id) {
return Err(EscalationError::DuplicateSessionRequest);
}
let session = VerifiedSession::new(
self.publication.trust_manifest_sha256,
self.credentials.trust_manifest.tls_ca_sha256,
&self.credentials.trust_manifest,
binding.clone(),
self.credentials.coordinator_protocol_credential(),
self.credentials.sandbox_protocol_credential(),
registration.session_capability.clone(),
)?;
let prefix = mailbox_session_name(session_id);
let mut binding_name = prefix.clone();
binding_name.push("-binding.postcard");
let mut registration_name = prefix;
registration_name.push("-registration.postcard");
let binding_identity = self.credential_files.publish(
&binding_name,
&lazaret_core::codec::encode(binding)
.map_err(|error| io::Error::other(error.to_string()))?,
)?;
let registration_identity = match self.credential_files.publish(
&registration_name,
&lazaret_core::codec::encode(registration)
.map_err(|error| io::Error::other(error.to_string()))?,
) {
Ok(identity) => identity,
Err(error) => {
self.credential_files
.remove_if_matches(&binding_name, binding_identity)?;
return Err(error.into());
}
};
self.sessions.insert(
session_id,
ProductionSessionAuthority {
session,
files: SessionAuthorityFiles {
binding_name,
binding_identity,
registration_name,
registration_identity,
},
},
);
Ok(())
}
fn disconnect(&mut self, session_id: SessionId) -> Result<(), EscalationError> {
self.coordinator.session_lost(session_id)?;
self.active.retain(|_, active| *active != session_id);
let _ = self.prompt_transport.finish_escalation(session_id);
if let Some(authority) = self.sessions.remove(&session_id) {
self.credential_files.remove_if_matches(
&authority.files.binding_name,
authority.files.binding_identity,
)?;
self.credential_files.remove_if_matches(
&authority.files.registration_name,
authority.files.registration_identity,
)?;
}
Ok(())
}
fn process_requests(&mut self) -> Result<(), EscalationError> {
for name in self.inbox.names()? {
if !is_ready_record(&name) || !name.as_bytes().starts_with(b"request-") {
continue;
}
let Ok((bytes, identity)) = self.inbox.read(&name, MAX_SIGNED_RECORD_BYTES) else {
continue;
};
let matched = self.sessions.values().find_map(|authority| {
authority
.session
.verify_request(&bytes)
.ok()
.map(|(request, _)| (authority.session.clone(), request))
});
self.inbox.remove_if_matches(&name, identity)?;
let Some((session, request)) = matched else {
continue;
};
let session_id = session.session_binding().session_id;
let request_id = request.request_id();
let auto = self
.configuration
.auto_approval_rules()
.iter()
.find_map(|rule| {
rule.authorize(&request)
.ok()
.map(|executable| (executable, request.command_argv().to_vec()))
});
let phase = if let Some((executable, command_argv)) = auto {
self.coordinator.receive_auto_approved(
session,
&bytes,
executable,
command_argv,
&mut self.ownership,
Instant::now(),
)
} else {
let phase = self.coordinator.receive_for_developer_approval(
session.clone(),
&bytes,
Instant::now(),
);
let Ok(phase) = phase else {
continue;
};
if let Err(error) = self.prompt_transport.queue_signed_request(session, &bytes) {
self.coordinator.session_lost(session_id)?;
return Err(error.into());
}
Ok(phase)
};
let phase = match phase {
Ok(phase) => phase,
Err(error) if error.is_record_refusal() => continue,
Err(error) => return Err(error),
};
self.record_phase(request_id, session_id, phase)?;
}
Ok(())
}
fn process_active(&mut self) -> Result<(), EscalationError> {
let active = self
.active
.iter()
.map(|(id, session)| (*id, *session))
.collect::<Vec<_>>();
for (request_id, session_id) in &active {
if let Some(decision) = self.prompt_transport.take_decision(*session_id)? {
let phase = self.coordinator.apply_developer_decision(
*request_id,
&decision.signed_decision,
&mut DirectDeveloperExecutableResolver::open()?,
&mut self.ownership,
)?;
self.record_phase(*request_id, *session_id, phase)?;
}
if !self.active.contains_key(request_id) {
continue;
}
let inbox_names = self.inbox.names()?;
let prefix = acknowledgment_prefix(*request_id);
for name in &inbox_names {
if !name.as_bytes().starts_with(&prefix) {
continue;
}
let Ok(bytes) = self.inbox.consume(name, MAX_SIGNED_RECORD_BYTES) else {
continue;
};
match self
.coordinator
.acknowledge(*request_id, &bytes, Instant::now())
{
Ok(()) => {}
Err(error) if error.is_record_refusal() => {}
Err(error) => return Err(error),
}
}
let cancellation = cancellation_file_name(*request_id);
if inbox_names.contains(&cancellation) {
let Ok(bytes) = self.inbox.consume(&cancellation, MAX_SIGNED_RECORD_BYTES) else {
continue;
};
match self.coordinator.cancel(*request_id, &bytes) {
Ok(phase) => self.record_phase(*request_id, *session_id, phase)?,
Err(error) if error.is_record_refusal() => {}
Err(error) => return Err(error),
}
continue;
}
if self.active.contains_key(request_id) {
let phase = self.coordinator.progress(*request_id, Instant::now())?;
self.record_phase(*request_id, *session_id, phase)?;
}
}
for request_id in self.coordinator.expire(Instant::now())? {
if let Some(session_id) = self.active.get(&request_id).copied() {
self.record_phase(
request_id,
session_id,
EscalationPhase::Delivered(EscalationOutcome::TimedOut),
)?;
}
}
Ok(())
}
fn record_phase(
&mut self,
request_id: RequestId,
session_id: SessionId,
phase: EscalationPhase,
) -> Result<(), EscalationError> {
match phase {
EscalationPhase::AwaitingApproval => {
self.active.insert(request_id, session_id);
}
EscalationPhase::Executing => {
self.active.insert(request_id, session_id);
self.prompt_transport.begin_host_execution(session_id)?;
}
EscalationPhase::Delivered(_) => {
self.active.remove(&request_id);
self.prompt_transport.finish_escalation(session_id)?;
}
}
Ok(())
}
}
fn acknowledgment_prefix(request_id: RequestId) -> Vec<u8> {
let example = acknowledgment_file_name(request_id, 0);
let bytes = example.as_bytes();
bytes[..bytes.len() - 26].to_vec()
}
fn validate_executable(executable: impl AsFd, invoking_uid: u32) -> io::Result<()> {
let stat = fstat(executable)?;
let file_type = FileType::from_raw_mode(stat.st_mode);

29
crates/lazaret-host-coordinator/src/production.rs
View file

@ -1,5 +1,5 @@
use std::cell::RefCell;
use std::ffi::OsString;
use std::ffi::{OsStr, OsString};
use std::fmt;
use std::fs::File;
use std::io::{Read, Write};
@ -16,7 +16,7 @@ use lazaret_core::records::{
Generation, GenerationIdentity, HostProcessOwnershipSlot, IpAddress, PreparedArtifact, Session,
};
use lazaret_ignored_workspace::git_common_directory;
use lazaret_linux::filesystem::{mount_identity, namespace_identity};
use lazaret_linux::filesystem::{RegularFileMailbox, mount_identity, namespace_identity};
use lazaret_linux::process::{
ProcessOwnershipContext, TargetCommand, child_broker, current_process_identity,
};
@ -707,6 +707,31 @@ impl RuntimeStaging {
Ok(())
}
pub(crate) fn escalation_mailboxes(
&self,
invoking_uid: u32,
) -> Result<
(RegularFileMailbox, RegularFileMailbox, RegularFileMailbox),
ProductionColdStartError,
> {
let credentials = RegularFileMailbox::openat_private(
&self.control,
OsStr::new("credentials"),
invoking_uid,
)
.map_err(failure)?;
let mailbox =
RegularFileMailbox::openat_private(&self.control, OsStr::new("mailbox"), invoking_uid)
.map_err(failure)?;
let requests =
RegularFileMailbox::openat_private(&mailbox, OsStr::new("requests"), invoking_uid)
.map_err(failure)?;
let results =
RegularFileMailbox::openat_private(&mailbox, OsStr::new("results"), invoking_uid)
.map_err(failure)?;
Ok((credentials, requests, results))
}
fn create(
runtime_domain: &ConfinedPathPlan,
container_id: ContainerId,

55
crates/lazaret-host-coordinator/src/terminal.rs
View file

@ -371,6 +371,56 @@ impl SessionPromptTransport {
.map_err(|_| TerminalError::PromptValidation)?
.remove(&session_id))
}
pub(crate) fn begin_host_execution(&self, session_id: SessionId) -> Result<(), TerminalError> {
self.transition_session(session_id, SessionPhase::HostExecuting)
}
pub(crate) fn finish_escalation(&self, session_id: SessionId) -> Result<(), TerminalError> {
self.pending
.lock()
.map_err(|_| TerminalError::PromptValidation)?
.remove(&session_id);
self.routed
.lock()
.map_err(|_| TerminalError::PromptValidation)?
.remove(&session_id);
let snapshot = self.store.snapshot()?;
let Some(session) = snapshot
.sessions
.iter()
.find(|session| session.session_id == session_id)
else {
return Err(TerminalError::UnknownSession);
};
match session.session_phase {
SessionPhase::Attached => return Ok(()),
SessionPhase::Prompting | SessionPhase::HostExecuting => {}
SessionPhase::New
| SessionPhase::Waiting
| SessionPhase::Registering
| SessionPhase::Closed
| SessionPhase::Failed => return Err(TerminalError::InactiveRoute),
}
self.transition_session(session_id, SessionPhase::Attached)
}
fn transition_session(
&self,
session_id: SessionId,
session_phase: SessionPhase,
) -> Result<(), TerminalError> {
let snapshot = self.store.snapshot()?;
let mut session = snapshot
.sessions
.into_iter()
.find(|session| session.session_id == session_id)
.ok_or(TerminalError::UnknownSession)?;
session.session_phase = session_phase;
self.store
.apply(&[DomainStoreMutation::PutSession(session)])?;
Ok(())
}
}
impl PromptRouteCancellation for SessionPromptTransport {
@ -1427,6 +1477,11 @@ impl SessionTerminalFactory for RunscSessionTerminalFactory {
"-internal-pid-file={}",
request.internal_pid_file.display()
))
.argument("-env")
.argument(format!(
"LAZARET_SESSION_ID={}",
hex_lower(request.session_id.as_bytes())
))
.argument(hex_lower(request.container_id.as_bytes()))
.argument(SANDBOX_INIT_PATH)
.argument(PRIVATE_SESSION_EXEC_ROLE)

106
crates/lazaret-host-coordinator/tests/escalation.rs
View file

@ -272,6 +272,31 @@ fn drive_terminal(
Err("terminal result deadline exhausted".into())
}
fn acknowledge_one_chunk(
mailboxes: &Mailboxes,
coordinator: &mut EscalationCoordinator,
request: &mut ClientRequest<'_>,
now: Instant,
) -> TestResult<Vec<u8>> {
for _ in 0..500 {
coordinator.progress(request.request_id(), now)?;
match request.poll()? {
ClientEvent::Output {
sequence, bytes, ..
} => {
let ack = mailboxes.consume_ack(request.request_id(), sequence)?;
coordinator.acknowledge(request.request_id(), &ack, now)?;
return Ok(bytes);
}
ClientEvent::Terminal(_) => {
return Err("request completed before partial output".into());
}
ClientEvent::Pending => std::thread::sleep(Duration::from_millis(1)),
}
}
Err("partial output deadline exhausted".into())
}
fn completed_auto_and_developer_requests_preserve_attribution() -> TestResult {
for (seed, auto) in [(10, true), (30, false)] {
let fixture = make_fixture(seed)?;
@ -336,6 +361,19 @@ fn completed_auto_and_developer_requests_preserve_attribution() -> TestResult {
ApprovalSource::Developer
}
);
assert!(
RegularFileMailbox::open(&mailboxes.outbox_path)?
.names()?
.contains(&lazaret_core::protocol::result_file_name(
request.request_id()
))
);
coordinator.session_lost(fixture.session.session_binding().session_id)?;
assert!(
RegularFileMailbox::open(&mailboxes.outbox_path)?
.names()?
.is_empty()
);
assert!(recover_process(ownership.last.unwrap().identity()).is_err());
}
Ok(())
@ -558,6 +596,73 @@ fn output_window_never_exceeds_eight_unacknowledged_chunks() -> TestResult {
Ok(())
}
fn interrupted_execution_commits_acknowledged_output() -> TestResult {
let fixture = make_fixture(115)?;
let mailboxes = Mailboxes::create("cancel-after-output")?;
let client = mailboxes.client(&fixture)?;
let argv = vec![
mailboxes.command("/usr/bin/head", "head")?,
b"-c".to_vec(),
b"1000000".to_vec(),
b"/dev/zero".to_vec(),
];
let mut request = client.request(argv.clone(), "cancel after output".into())?;
let signed = mailboxes.consume_request(request.request_id())?;
let now = Instant::now();
let mut coordinator =
mailboxes.coordinator(&fixture, Duration::from_secs(5), Duration::from_secs(2))?;
coordinator.receive_auto_approved(
fixture.session.clone(),
&signed,
executable(&argv)?,
argv,
&mut RecordingOwnership::default(),
now,
)?;
assert!(!acknowledge_one_chunk(&mailboxes, &mut coordinator, &mut request, now)?.is_empty());
request.cancel()?;
let cancellation = RegularFileMailbox::open(&mailboxes.inbox_path)?.consume(
&lazaret_core::protocol::cancellation_file_name(request.request_id()),
MAX_SIGNED_RECORD_BYTES,
)?;
coordinator.cancel(request.request_id(), &cancellation)?;
assert!(
matches!(request.poll()?, ClientEvent::Terminal(result) if result.outcome == EscalationOutcome::Cancelled)
);
let fixture = make_fixture(116)?;
let mailboxes = Mailboxes::create("timeout-after-output")?;
let client = mailboxes.client(&fixture)?;
let argv = vec![
mailboxes.command("/usr/bin/head", "head")?,
b"-c".to_vec(),
b"1000000".to_vec(),
b"/dev/zero".to_vec(),
];
let mut request = client.request(argv.clone(), "timeout after output".into())?;
let signed = mailboxes.consume_request(request.request_id())?;
let start = Instant::now();
let mut coordinator =
mailboxes.coordinator(&fixture, Duration::from_millis(10), Duration::from_secs(2))?;
coordinator.receive_auto_approved(
fixture.session.clone(),
&signed,
executable(&argv)?,
argv,
&mut RecordingOwnership::default(),
start,
)?;
assert!(!acknowledge_one_chunk(&mailboxes, &mut coordinator, &mut request, start)?.is_empty());
assert_eq!(
coordinator.expire(start + Duration::from_millis(20))?,
vec![request.request_id()]
);
assert!(
matches!(request.poll()?, ClientEvent::Terminal(result) if result.outcome == EscalationOutcome::TimedOut)
);
Ok(())
}
fn replay_and_session_tuple_mismatch_refuse() -> TestResult {
let fixture = make_fixture(120)?;
let mailboxes = Mailboxes::create("replay")?;
@ -767,6 +872,7 @@ fn run() -> TestResult {
denial_failure_and_request_timeout_publish_terminal_outcomes()?;
cancellation_session_loss_and_ack_timeout_reap_exact_children()?;
output_window_never_exceeds_eight_unacknowledged_chunks()?;
interrupted_execution_commits_acknowledged_output()?;
replay_and_session_tuple_mismatch_refuse()?;
signed_cancellation_decision_and_result_tuple_mismatches_refuse()?;
exact_descriptor_launcher_survives_path_replacement()?;