Add confined host escalation transport

crates/lazaret-host-coordinator/src/escalation.rs

@@ -0,0 +1,941 @@
+use std::cell::RefCell;
+use std::collections::BTreeMap;
+use std::ffi::{OsStr, OsString};
+use std::fmt;
+use std::io::{self, Read};
+use std::os::fd::{AsFd, OwnedFd};
+use std::os::unix::ffi::{OsStrExt, OsStringExt};
+use std::rc::Rc;
+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,
+};
+use lazaret_core::records::{
+    DeveloperDecisionKind, EscalationOutcome, GenerationIdentity, HostProcessOwnershipSlot,
+    MailboxRequest, MailboxResult, MailboxResultChunk, ProcessOwnershipCommitAck, ProtocolVersion,
+    ResultStream, StagedProcessOwnership,
+};
+use lazaret_linux::filesystem::{MailboxEntryIdentity, RegularFileMailbox};
+use lazaret_linux::process::{
+    CapturedProcess, ProcessOwnershipContext, TargetCommand, child_broker, current_process_identity,
+};
+use rustix::fs::{FileType, Mode, OFlags, ResolveFlags, fchmod, fstat, mkdirat, openat, openat2};
+use rustix::process::Signal;
+use sha2::{Digest, Sha256};
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ApprovalSource {
+    AutoRule,
+    Developer,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ApprovalAttribution {
+    pub source: ApprovalSource,
+    pub command_argv: Vec<Vec<u8>>,
+}
+
+pub struct GenerationMailbox {
+    generation_directory: RegularFileMailbox,
+    inbox: RegularFileMailbox,
+    outbox: RegularFileMailbox,
+}
+
+impl GenerationMailbox {
+    /// Creates the fixed generation path beneath an already-confined runtime Domain root.
+    ///
+    /// # Errors
+    /// Refuses links, changed ownership or mode, existing identity, or incomplete publication.
+    pub fn create(
+        runtime_domain_root: impl AsFd,
+        generation: &GenerationIdentity,
+        invoking_uid: u32,
+    ) -> Result<Self, EscalationError> {
+        let mailbox =
+            create_private_directory(runtime_domain_root, OsStr::new("mailbox"), invoking_uid)?;
+        let generation_name =
+            lazaret_core::protocol::mailbox_generation_name(generation.generation_id.get());
+        let generation_directory =
+            create_private_directory(&mailbox, &generation_name, invoking_uid)?;
+        let identity = lazaret_core::codec::encode(generation)
+            .map_err(|error| io::Error::other(error.to_string()))?;
+        generation_directory.publish(OsStr::new("generation-identity.postcard"), &identity)?;
+        let inbox =
+            create_private_directory(&generation_directory, OsStr::new("inbox"), invoking_uid)?;
+        let outbox =
+            create_private_directory(&generation_directory, OsStr::new("outbox"), invoking_uid)?;
+        Ok(Self {
+            generation_directory,
+            inbox,
+            outbox,
+        })
+    }
+
+    #[must_use]
+    pub const fn inbox(&self) -> &RegularFileMailbox {
+        &self.inbox
+    }
+
+    #[must_use]
+    pub const fn outbox(&self) -> &RegularFileMailbox {
+        &self.outbox
+    }
+
+    #[must_use]
+    pub fn into_directories(self) -> (RegularFileMailbox, RegularFileMailbox) {
+        let _ = self.generation_directory;
+        (self.inbox, self.outbox)
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum EscalationPhase {
+    AwaitingApproval,
+    Executing,
+    Delivered(EscalationOutcome),
+}
+
+pub trait DeveloperExecutableResolver {
+    /// Opens the executable named by element zero and returns authority over that exact object.
+    ///
+    /// # Errors
+    /// Refuses an argv vector that cannot be resolved to one approved executable descriptor.
+    fn resolve(&mut self, command_argv: &[Vec<u8>]) -> io::Result<OwnedFd>;
+}
+
+pub struct DirectDeveloperExecutableResolver {
+    root: OwnedFd,
+}
+
+fn create_private_directory(
+    parent: impl AsFd,
+    name: &OsStr,
+    invoking_uid: u32,
+) -> io::Result<RegularFileMailbox> {
+    let created = match mkdirat(parent.as_fd(), name, Mode::RUSR | Mode::WUSR | Mode::XUSR) {
+        Ok(()) => true,
+        Err(rustix::io::Errno::EXIST) => false,
+        Err(error) => return Err(error.into()),
+    };
+    let directory = openat(
+        parent.as_fd(),
+        name,
+        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+        Mode::empty(),
+    )?;
+    if created {
+        fchmod(&directory, Mode::RUSR | Mode::WUSR | Mode::XUSR)?;
+    }
+    drop(directory);
+    RegularFileMailbox::openat_private(parent, name, invoking_uid)
+}
+
+impl DirectDeveloperExecutableResolver {
+    /// # Errors
+    /// Returns an error when host root authority cannot be retained without following a link.
+    pub fn open() -> io::Result<Self> {
+        Ok(Self {
+            root: openat(
+                rustix::fs::CWD,
+                "/",
+                OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+                Mode::empty(),
+            )?,
+        })
+    }
+}
+
+impl DeveloperExecutableResolver for DirectDeveloperExecutableResolver {
+    fn resolve(&mut self, command_argv: &[Vec<u8>]) -> io::Result<OwnedFd> {
+        let executable = command_argv
+            .first()
+            .ok_or_else(|| io::Error::other("approved command vector is empty"))?;
+        if executable.contains(&0) || !executable.starts_with(b"/") {
+            return Err(io::Error::other(
+                "approved executable is not a NUL-free absolute path",
+            ));
+        }
+        let relative = std::ffi::OsStr::from_bytes(&executable[1..]);
+        openat2(
+            &self.root,
+            relative,
+            OFlags::RDONLY | OFlags::NONBLOCK | OFlags::CLOEXEC,
+            Mode::empty(),
+            ResolveFlags::BENEATH | ResolveFlags::NO_MAGICLINKS | ResolveFlags::NO_SYMLINKS,
+        )
+        .map_err(Into::into)
+    }
+}
+
+pub trait ExecutionOwnershipRecorder {
+    /// # Errors
+    /// Refuses when exact staged ownership cannot be retained.
+    fn stage(&mut self, ownership: StagedProcessOwnership) -> io::Result<()>;
+
+    /// # Errors
+    /// Refuses missing or mismatched exact ownership acknowledgement.
+    fn acknowledge(
+        &mut self,
+        ownership: StagedProcessOwnership,
+    ) -> io::Result<ProcessOwnershipCommitAck>;
+}
+
+#[derive(Debug)]
+pub enum EscalationError {
+    Io(io::Error),
+    Protocol(ProtocolValidationError),
+    ProtocolCodec(ProtocolCodecError),
+    Window(ResultWindowError),
+    DuplicateSessionRequest,
+    UnknownRequest,
+    InvalidPhase,
+    InvalidApproval,
+}
+
+impl fmt::Display for EscalationError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Io(error) => error.fmt(formatter),
+            Self::Protocol(error) => error.fmt(formatter),
+            Self::ProtocolCodec(error) => error.fmt(formatter),
+            Self::Window(error) => error.fmt(formatter),
+            Self::DuplicateSessionRequest => {
+                formatter.write_str("session already has an active escalation request")
+            }
+            Self::UnknownRequest => formatter.write_str("escalation request is unknown"),
+            Self::InvalidPhase => formatter.write_str("operation is invalid in the request phase"),
+            Self::InvalidApproval => {
+                formatter.write_str("developer approval does not authorize an executable vector")
+            }
+        }
+    }
+}
+
+impl std::error::Error for EscalationError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            Self::Io(error) => Some(error),
+            Self::Protocol(error) => Some(error),
+            Self::ProtocolCodec(error) => Some(error),
+            Self::Window(error) => Some(error),
+            _ => None,
+        }
+    }
+}
+
+impl From<io::Error> for EscalationError {
+    fn from(error: io::Error) -> Self {
+        Self::Io(error)
+    }
+}
+
+impl From<ProtocolValidationError> for EscalationError {
+    fn from(error: ProtocolValidationError) -> Self {
+        Self::Protocol(error)
+    }
+}
+
+impl From<ProtocolCodecError> for EscalationError {
+    fn from(error: ProtocolCodecError) -> Self {
+        Self::ProtocolCodec(error)
+    }
+}
+
+impl From<ResultWindowError> for EscalationError {
+    fn from(error: ResultWindowError) -> Self {
+        Self::Window(error)
+    }
+}
+
+struct RequestAuthority {
+    session: VerifiedSession,
+    request: MailboxRequest,
+    digest: RequestDigest,
+    admitted_at: Instant,
+}
+
+struct PublishedChunk {
+    name: OsString,
+    identity: MailboxEntryIdentity,
+}
+
+struct Execution {
+    authority: RequestAuthority,
+    process: CapturedProcess,
+    window: ResultSenderWindow,
+    chunks: BTreeMap<u64, PublishedChunk>,
+    stdout: Sha256,
+    stderr: Sha256,
+    stdout_eof: bool,
+    stderr_eof: bool,
+    acknowledgment_deadline: Option<Instant>,
+}
+
+enum ActiveRequest {
+    AwaitingApproval(Box<RequestAuthority>),
+    Executing(Box<Execution>),
+}
+
+struct PublishedResult {
+    session_id: SessionId,
+    outcome: EscalationOutcome,
+    name: OsString,
+    identity: MailboxEntryIdentity,
+}
+
+pub struct EscalationCoordinator {
+    coordinator_key: SigningKey,
+    outbox: RegularFileMailbox,
+    ownership: ProcessOwnershipContext,
+    invoking_uid: u32,
+    request_timeout: Duration,
+    acknowledgment_timeout: Duration,
+    replay: RequestReplayGuard,
+    active_sessions: BTreeMap<SessionId, RequestId>,
+    active: BTreeMap<RequestId, ActiveRequest>,
+    approvals: BTreeMap<RequestId, ApprovalAttribution>,
+    results: BTreeMap<RequestId, PublishedResult>,
+}
+
+impl EscalationCoordinator {
+    #[must_use]
+    pub fn new(
+        coordinator_key: SigningKey,
+        outbox: RegularFileMailbox,
+        ownership: ProcessOwnershipContext,
+        invoking_uid: u32,
+        request_timeout: Duration,
+        acknowledgment_timeout: Duration,
+    ) -> Self {
+        Self {
+            coordinator_key,
+            outbox,
+            ownership,
+            invoking_uid,
+            request_timeout,
+            acknowledgment_timeout,
+            replay: RequestReplayGuard::default(),
+            active_sessions: BTreeMap::new(),
+            active: BTreeMap::new(),
+            approvals: BTreeMap::new(),
+            results: BTreeMap::new(),
+        }
+    }
+
+    /// Admits one exact signed request and routes matching configuration rules automatically.
+    ///
+    /// # Errors
+    /// Refuses replay, stale tuples, a second request for one session, or launch failure.
+    pub fn receive_request<R: ExecutionOwnershipRecorder>(
+        &mut self,
+        session: VerifiedSession,
+        signed_request: &[u8],
+        configuration: &ResolvedDomainConfiguration,
+        recorder: &mut R,
+        now: Instant,
+    ) -> Result<EscalationPhase, EscalationError> {
+        let authority = self.admit(session, signed_request, now)?;
+        let request_id = authority.request.request_id();
+        for rule in configuration.auto_approval_rules() {
+            if let Ok(executable) = rule.authorize(&authority.request) {
+                let command_argv = authority.request.command_argv().to_vec();
+                return self.start_execution(
+                    authority,
+                    ApprovalSource::AutoRule,
+                    executable,
+                    command_argv,
+                    recorder,
+                );
+            }
+        }
+        self.active.insert(
+            request_id,
+            ActiveRequest::AwaitingApproval(Box::new(authority)),
+        );
+        Ok(EscalationPhase::AwaitingApproval)
+    }
+
+    /// Admits a request into the developer-decision path without composing prompt UI.
+    ///
+    /// # Errors
+    /// Refuses replay, stale tuples, or a second request for one session.
+    pub fn receive_for_developer_approval(
+        &mut self,
+        session: VerifiedSession,
+        signed_request: &[u8],
+        now: Instant,
+    ) -> Result<EscalationPhase, EscalationError> {
+        let authority = self.admit(session, signed_request, now)?;
+        let request_id = authority.request.request_id();
+        self.active.insert(
+            request_id,
+            ActiveRequest::AwaitingApproval(Box::new(authority)),
+        );
+        Ok(EscalationPhase::AwaitingApproval)
+    }
+
+    /// Admits an exact request after an external rule matcher supplies its already-open descriptor.
+    ///
+    /// # Errors
+    /// Refuses changed argv, replay, stale tuples, duplicate session activity, or launch failure.
+    pub fn receive_auto_approved<R: ExecutionOwnershipRecorder>(
+        &mut self,
+        session: VerifiedSession,
+        signed_request: &[u8],
+        executable: OwnedFd,
+        command_argv: Vec<Vec<u8>>,
+        recorder: &mut R,
+        now: Instant,
+    ) -> Result<EscalationPhase, EscalationError> {
+        let authority = self.admit(session, signed_request, now)?;
+        if authority.request.command_argv() != command_argv {
+            let session_id = authority.request.session_binding().session_id;
+            self.active_sessions.remove(&session_id);
+            return Err(EscalationError::InvalidApproval);
+        }
+        self.start_execution(
+            authority,
+            ApprovalSource::AutoRule,
+            executable,
+            command_argv,
+            recorder,
+        )
+    }
+
+    fn admit(
+        &mut self,
+        session: VerifiedSession,
+        signed_request: &[u8],
+        now: Instant,
+    ) -> Result<RequestAuthority, EscalationError> {
+        let (request, _) = session.verify_request(signed_request)?;
+        let session_id = request.session_binding().session_id;
+        if self.active_sessions.contains_key(&session_id) {
+            return Err(EscalationError::DuplicateSessionRequest);
+        }
+        let (request, digest) = self.replay.admit(&session, signed_request)?;
+        let request_id = request.request_id();
+        self.active_sessions.insert(session_id, request_id);
+        Ok(RequestAuthority {
+            session,
+            request,
+            digest,
+            admitted_at: now,
+        })
+    }
+
+    #[must_use]
+    pub fn approval_source(&self, request_id: RequestId) -> Option<ApprovalSource> {
+        self.approvals
+            .get(&request_id)
+            .map(|attribution| attribution.source)
+    }
+
+    #[must_use]
+    pub fn approval_attribution(&self, request_id: RequestId) -> Option<&ApprovalAttribution> {
+        self.approvals.get(&request_id)
+    }
+
+    /// Applies one exact Session Anchor decision without providing any prompt or PTY UI.
+    ///
+    /// # Errors
+    /// Refuses stale, mismatched, unsigned, or phase-inconsistent decisions.
+    pub fn apply_developer_decision<Resolver, Recorder>(
+        &mut self,
+        request_id: RequestId,
+        signed_decision: &[u8],
+        resolver: &mut Resolver,
+        recorder: &mut Recorder,
+    ) -> Result<EscalationPhase, EscalationError>
+    where
+        Resolver: DeveloperExecutableResolver,
+        Recorder: ExecutionOwnershipRecorder,
+    {
+        let Some(ActiveRequest::AwaitingApproval(authority)) = self.active.get(&request_id) else {
+            return Err(EscalationError::InvalidPhase);
+        };
+        let decision = authority.session.verify_developer_decision(
+            signed_decision,
+            request_id,
+            authority.digest,
+        )?;
+        let executable = match decision.decision {
+            DeveloperDecisionKind::Deny => None,
+            DeveloperDecisionKind::Approve => {
+                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)
+            }
+        };
+        let Some(ActiveRequest::AwaitingApproval(authority)) = self.active.remove(&request_id)
+        else {
+            return Err(EscalationError::InvalidPhase);
+        };
+        match decision.decision {
+            DeveloperDecisionKind::Deny => {
+                self.finish(&authority, EscalationOutcome::Denied, None)?;
+                Ok(EscalationPhase::Delivered(EscalationOutcome::Denied))
+            }
+            DeveloperDecisionKind::Approve => self.start_execution(
+                *authority,
+                ApprovalSource::Developer,
+                executable.ok_or(EscalationError::InvalidApproval)?,
+                decision.approved_command_argv,
+                recorder,
+            ),
+        }
+    }
+
+    fn start_execution<R: ExecutionOwnershipRecorder>(
+        &mut self,
+        authority: RequestAuthority,
+        approval_source: ApprovalSource,
+        executable: OwnedFd,
+        command_argv: Vec<Vec<u8>>,
+        recorder: &mut R,
+    ) -> Result<EscalationPhase, EscalationError> {
+        let request_id = authority.request.request_id();
+        self.approvals.insert(
+            request_id,
+            ApprovalAttribution {
+                source: approval_source,
+                command_argv: command_argv.clone(),
+            },
+        );
+        let setup = (|| {
+            validate_executable(executable.as_fd(), self.invoking_uid)?;
+            let arguments = command_argv
+                .into_iter()
+                .skip(1)
+                .map(OsString::from_vec)
+                .collect::<Vec<_>>();
+            let target = TargetCommand::from_executable(executable, arguments)?;
+            let parent = current_process_identity()?;
+            Ok::<_, io::Error>((target, parent))
+        })();
+        let Ok((target, parent)) = setup else {
+            self.finish(&authority, EscalationOutcome::Failed, None)?;
+            return Ok(EscalationPhase::Delivered(EscalationOutcome::Failed));
+        };
+        let recorder = Rc::new(RefCell::new(recorder));
+        let stage_recorder = Rc::clone(&recorder);
+        let process = child_broker().acquire_captured(
+            parent,
+            ProcessOwnershipContext {
+                slot: HostProcessOwnershipSlot::EscalationHostProcess(request_id),
+                ..self.ownership
+            },
+            &target,
+            move |staged| stage_recorder.borrow_mut().stage(staged),
+            move |staged| recorder.borrow_mut().acknowledge(staged),
+        );
+        let Ok(process) = process else {
+            self.finish(&authority, EscalationOutcome::Failed, None)?;
+            return Ok(EscalationPhase::Delivered(EscalationOutcome::Failed));
+        };
+        self.active.insert(
+            request_id,
+            ActiveRequest::Executing(Box::new(Execution {
+                authority,
+                process,
+                window: ResultSenderWindow::default(),
+                chunks: BTreeMap::new(),
+                stdout: Sha256::new(),
+                stderr: Sha256::new(),
+                stdout_eof: false,
+                stderr_eof: false,
+                acknowledgment_deadline: None,
+            })),
+        );
+        Ok(EscalationPhase::Executing)
+    }
+
+    /// Advances nonblocking output while preserving the eight-chunk kernel backpressure bound.
+    ///
+    /// # Errors
+    /// Returns an error when exact output publication, process observation, or reaping fails.
+    pub fn progress(
+        &mut self,
+        request_id: RequestId,
+        now: Instant,
+    ) -> Result<EscalationPhase, EscalationError> {
+        let Some(active) = self.active.remove(&request_id) else {
+            return self
+                .results
+                .get(&request_id)
+                .map(|result| EscalationPhase::Delivered(result.outcome))
+                .ok_or(EscalationError::UnknownRequest);
+        };
+        let ActiveRequest::Executing(mut execution) = active else {
+            self.active.insert(request_id, active);
+            return Ok(EscalationPhase::AwaitingApproval);
+        };
+        let progress = self.read_available(&mut execution, now);
+        if progress.is_err() {
+            let authority = self.terminate(*execution)?;
+            self.finish(&authority, EscalationOutcome::Failed, None)?;
+            return Ok(EscalationPhase::Delivered(EscalationOutcome::Failed));
+        }
+        if execution.process.process().has_exited()?
+            && execution.stdout_eof
+            && execution.stderr_eof
+            && execution.window.outstanding() == 0
+        {
+            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 status = child_broker().reap_status(execution.process.into_process())?;
+            self.finish_with_streams(
+                &authority,
+                EscalationOutcome::Completed,
+                status.exit_code,
+                chunk_count,
+                stdout,
+                stderr,
+            )?;
+            return Ok(EscalationPhase::Delivered(EscalationOutcome::Completed));
+        }
+        self.active
+            .insert(request_id, ActiveRequest::Executing(execution));
+        Ok(EscalationPhase::Executing)
+    }
+
+    fn read_available(
+        &mut self,
+        execution: &mut Execution,
+        now: Instant,
+    ) -> Result<(), EscalationError> {
+        while execution.window.outstanding() < 8 {
+            let mut made_progress = false;
+            for stream in [ResultStream::Stdout, ResultStream::Stderr] {
+                if execution.window.outstanding() >= 8 {
+                    break;
+                }
+                let eof = match stream {
+                    ResultStream::Stdout => &mut execution.stdout_eof,
+                    ResultStream::Stderr => &mut execution.stderr_eof,
+                };
+                if *eof {
+                    continue;
+                }
+                let (stdout, stderr) = execution.process.streams();
+                let reader = match stream {
+                    ResultStream::Stdout => stdout,
+                    ResultStream::Stderr => stderr,
+                };
+                let mut bytes = vec![0_u8; MAX_RESULT_CHUNK_BYTES];
+                match reader.read(&mut bytes) {
+                    Ok(0) => {
+                        *eof = true;
+                        made_progress = true;
+                    }
+                    Ok(length) => {
+                        bytes.truncate(length);
+                        let sequence = execution.window.reserve(length)?;
+                        let chunk = MailboxResultChunk {
+                            protocol_version: ProtocolVersion::V1,
+                            session_binding: execution.authority.session.session_binding().clone(),
+                            request_id: execution.authority.request.request_id(),
+                            request_sha256: execution.authority.digest,
+                            coordinator_credential_id: execution
+                                .authority
+                                .session
+                                .coordinator_credential_id(),
+                            sequence,
+                            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 });
+                        if execution.acknowledgment_deadline.is_none() {
+                            execution.acknowledgment_deadline =
+                                Some(now + self.acknowledgment_timeout);
+                        }
+                        made_progress = true;
+                    }
+                    Err(error) if error.kind() == io::ErrorKind::WouldBlock => {}
+                    Err(error) => return Err(error.into()),
+                }
+            }
+            if !made_progress {
+                break;
+            }
+        }
+        Ok(())
+    }
+
+    /// Accepts a highest-contiguous signed acknowledgement and removes only matching chunks.
+    ///
+    /// # Errors
+    /// Refuses tuple mismatch, replay, regression, or acknowledgement beyond sent output.
+    pub fn acknowledge(
+        &mut self,
+        request_id: RequestId,
+        signed_acknowledgment: &[u8],
+        now: Instant,
+    ) -> Result<(), EscalationError> {
+        let Some(ActiveRequest::Executing(execution)) = self.active.get_mut(&request_id) else {
+            return Err(EscalationError::InvalidPhase);
+        };
+        let ack = execution.authority.session.verify_result_ack(
+            signed_acknowledgment,
+            request_id,
+            execution.authority.digest,
+        )?;
+        execution
+            .window
+            .acknowledge(ack.highest_contiguous_sequence)?;
+        let acknowledged = execution
+            .chunks
+            .range(..=ack.highest_contiguous_sequence)
+            .map(|(sequence, _)| *sequence)
+            .collect::<Vec<_>>();
+        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);
+            }
+        }
+        execution.acknowledgment_deadline = if execution.window.outstanding() == 0 {
+            None
+        } else {
+            Some(now + self.acknowledgment_timeout)
+        };
+        Ok(())
+    }
+
+    /// Cancels an exact live request after verifying the sandbox signature and complete tuple.
+    ///
+    /// # Errors
+    /// Refuses unknown, terminal, unsigned, or mismatched cancellations.
+    pub fn cancel(
+        &mut self,
+        request_id: RequestId,
+        signed_cancellation: &[u8],
+    ) -> Result<EscalationPhase, EscalationError> {
+        let Some(active) = self.active.get(&request_id) else {
+            return Err(EscalationError::UnknownRequest);
+        };
+        match active {
+            ActiveRequest::AwaitingApproval(authority) => {
+                authority
+                    .session
+                    .verify_cancellation(signed_cancellation, request_id)?;
+            }
+            ActiveRequest::Executing(execution) => {
+                execution
+                    .authority
+                    .session
+                    .verify_cancellation(signed_cancellation, request_id)?;
+            }
+        }
+        let active = self
+            .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)?;
+        Ok(EscalationPhase::Delivered(EscalationOutcome::Cancelled))
+    }
+
+    /// Enforces request and acknowledgement deadlines with exact pidfd termination and reaping.
+    ///
+    /// # Errors
+    /// Returns an error when exact cleanup or terminal publication fails.
+    pub fn expire(&mut self, now: Instant) -> Result<Vec<RequestId>, EscalationError> {
+        let expired = self
+            .active
+            .iter()
+            .filter_map(|(request_id, active)| {
+                let expired = match active {
+                    ActiveRequest::AwaitingApproval(authority) => {
+                        now.saturating_duration_since(authority.admitted_at) >= self.request_timeout
+                    }
+                    ActiveRequest::Executing(execution) => {
+                        now.saturating_duration_since(execution.authority.admitted_at)
+                            >= self.request_timeout
+                            || execution
+                                .acknowledgment_deadline
+                                .is_some_and(|deadline| now >= deadline)
+                    }
+                };
+                expired.then_some(*request_id)
+            })
+            .collect::<Vec<_>>();
+        for request_id in &expired {
+            let active = self
+                .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)?;
+        }
+        Ok(expired)
+    }
+
+    /// Removes all exact result files and exact-child authority for a disconnected session.
+    ///
+    /// # Errors
+    /// Returns an error when exact cleanup cannot complete.
+    pub fn session_lost(&mut self, session_id: SessionId) -> Result<(), EscalationError> {
+        if let Some(request_id) = self.active_sessions.remove(&session_id)
+            && let Some(active) = self.active.remove(&request_id)
+        {
+            match active {
+                ActiveRequest::AwaitingApproval(_) => {}
+                ActiveRequest::Executing(execution) => {
+                    self.terminate(*execution)?;
+                }
+            }
+        }
+        let result_ids = self
+            .results
+            .iter()
+            .filter_map(|(request_id, result)| {
+                (result.session_id == session_id).then_some(*request_id)
+            })
+            .collect::<Vec<_>>();
+        for request_id in result_ids {
+            if let Some(result) = self.results.remove(&request_id) {
+                self.outbox
+                    .remove_if_matches(&result.name, result.identity)?;
+            }
+        }
+        Ok(())
+    }
+
+    fn terminate(&mut self, execution: Execution) -> Result<RequestAuthority, EscalationError> {
+        execution.process.process().send_signal(Signal::KILL)?;
+        let authority = execution.authority;
+        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)
+    }
+
+    fn finish(
+        &mut self,
+        authority: &RequestAuthority,
+        outcome: EscalationOutcome,
+        exit_code: Option<i32>,
+    ) -> Result<(), EscalationError> {
+        self.finish_with_streams(
+            authority,
+            outcome,
+            exit_code,
+            0,
+            StreamDigest::new(Sha256::digest([]).into()),
+            StreamDigest::new(Sha256::digest([]).into()),
+        )
+    }
+
+    fn finish_with_streams(
+        &mut self,
+        authority: &RequestAuthority,
+        outcome: EscalationOutcome,
+        exit_code: Option<i32>,
+        chunk_count: u64,
+        stdout_sha256: StreamDigest,
+        stderr_sha256: StreamDigest,
+    ) -> Result<(), EscalationError> {
+        let request_id = authority.request.request_id();
+        let session_id = authority.request.session_binding().session_id;
+        let result = MailboxResult {
+            protocol_version: ProtocolVersion::V1,
+            session_binding: authority.session.session_binding().clone(),
+            request_id,
+            request_sha256: authority.digest,
+            coordinator_credential_id: authority.session.coordinator_credential_id(),
+            outcome,
+            exit_code,
+            chunk_count,
+            stdout_sha256,
+            stderr_sha256,
+        };
+        let encoded = encode_signed(&result, &self.coordinator_key)?;
+        let name = result_file_name(request_id);
+        let identity = self.outbox.publish(&name, &encoded)?;
+        self.active_sessions.remove(&session_id);
+        self.results.insert(
+            request_id,
+            PublishedResult {
+                session_id,
+                outcome,
+                name,
+                identity,
+            },
+        );
+        Ok(())
+    }
+}
+
+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);
+    let mode = Mode::from_raw_mode(stat.st_mode);
+    if !file_type.is_file()
+        || stat.st_nlink != 1
+        || (stat.st_uid != 0 && stat.st_uid != invoking_uid)
+        || mode.intersects(Mode::WGRP | Mode::WOTH)
+        || !mode.contains(Mode::XUSR)
+    {
+        return Err(io::Error::other(
+            "approved executable changed type, links, ownership, mode, or executability",
+        ));
+    }
+    Ok(())
+}
+
+#[derive(Default)]
+pub struct LocalExecutionOwnership {
+    staged: Vec<StagedProcessOwnership>,
+}
+
+impl ExecutionOwnershipRecorder for LocalExecutionOwnership {
+    fn stage(&mut self, ownership: StagedProcessOwnership) -> io::Result<()> {
+        if self.staged.contains(&ownership) {
+            Err(io::Error::other("execution ownership was already staged"))
+        } else {
+            self.staged.push(ownership);
+            Ok(())
+        }
+    }
+
+    fn acknowledge(
+        &mut self,
+        ownership: StagedProcessOwnership,
+    ) -> io::Result<ProcessOwnershipCommitAck> {
+        if let Some(index) = self.staged.iter().position(|staged| *staged == ownership) {
+            self.staged.swap_remove(index);
+            Ok(ProcessOwnershipCommitAck::v1(ownership))
+        } else {
+            Err(io::Error::other("execution ownership is not staged"))
+        }
+    }
+}