Complete authenticated interactive prompt routing

This commit is contained in:
ms2k 2026-07-17 07:20:21 +09:00
commit 46b52b4b43
9 changed files with 595 additions and 96 deletions

12
Cargo.lock generated
View file

@ -1214,10 +1214,10 @@ dependencies = [
"lazaret-host-coordinator", "lazaret-host-coordinator",
"lazaret-linux", "lazaret-linux",
"lazaret-runtime-supervisor", "lazaret-runtime-supervisor",
"nix 0.31.3",
"rustix", "rustix",
"rustls", "rustls",
"sha2 0.11.0", "sha2 0.11.0",
"signal-hook",
"tempfile", "tempfile",
] ]
@ -2240,16 +2240,6 @@ version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba"
[[package]]
name = "signal-hook"
version = "0.3.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d881a16cf4426aa584979d30bd82cb33429027e42122b169753d6ef1085ed6e2"
dependencies = [
"libc",
"signal-hook-registry",
]
[[package]] [[package]]
name = "signal-hook-registry" name = "signal-hook-registry"
version = "1.4.8" version = "1.4.8"

View file

@ -34,11 +34,10 @@ postcard = { version = "=1.1.3", default-features = false, features = ["use-std"
rcgen = { version = "=0.14.8", default-features = false, features = ["aws_lc_rs", "x509-parser"] } rcgen = { version = "=0.14.8", default-features = false, features = ["aws_lc_rs", "x509-parser"] }
redb = "=4.1.0" redb = "=4.1.0"
libc = "=0.2.186" libc = "=0.2.186"
nix = { version = "=0.31.3", features = ["fs"] } nix = { version = "=0.31.3", features = ["fs", "signal"] }
rustls = { version = "=0.23.42", default-features = false, features = ["aws_lc_rs", "std"] } rustls = { version = "=0.23.42", default-features = false, features = ["aws_lc_rs", "std"] }
serde = { version = "=1.0.228", features = ["derive"] } serde = { version = "=1.0.228", features = ["derive"] }
serde_json = "=1.0.150" serde_json = "=1.0.150"
signal-hook = "=0.3.18"
rustix = { version = "=1.1.4", features = ["event", "fs", "mount", "net", "pipe", "process", "pty", "stdio", "termios", "thread", "time"] } rustix = { version = "=1.1.4", features = ["event", "fs", "mount", "net", "pipe", "process", "pty", "stdio", "termios", "thread", "time"] }
sha2 = "=0.11.0" sha2 = "=0.11.0"
tar = "=0.4.46" tar = "=0.4.46"

View file

@ -19,7 +19,7 @@ lazaret-runtime-supervisor = { version = "=0.0.0", path = "../lazaret-runtime-su
rustix.workspace = true rustix.workspace = true
rustls.workspace = true rustls.workspace = true
sha2.workspace = true sha2.workspace = true
signal-hook.workspace = true nix.workspace = true
[dev-dependencies] [dev-dependencies]
tempfile = "=3.27.0" tempfile = "=3.27.0"

View file

@ -1,8 +1,6 @@
use std::fs::{File, OpenOptions}; use std::fs::{File, OpenOptions};
use std::io::{self, Read, Write}; use std::io::{self, Read, Write};
use std::num::NonZeroU16; use std::num::NonZeroU16;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration; use std::time::Duration;
use lazaret_core::records::{ use lazaret_core::records::{
@ -11,10 +9,10 @@ use lazaret_core::records::{
SessionClose, SessionExitOutcome, SessionTerminalEvent, SessionTerminalMessage, TerminalInput, SessionClose, SessionExitOutcome, SessionTerminalEvent, SessionTerminalMessage, TerminalInput,
TerminalPrompt, TerminalResize, TerminalSignal, TerminalSignalKind, TerminalPrompt, TerminalResize, TerminalSignal, TerminalSignalKind,
}; };
use nix::sys::signal::{SigSet, SigmaskHow, Signal};
use nix::sys::signalfd::{SfdFlags, SignalFd};
use rustix::fs::{OFlags, fcntl_getfl, fcntl_setfl}; use rustix::fs::{OFlags, fcntl_getfl, fcntl_setfl};
use rustix::termios::{LocalModes, OptionalActions, Termios, tcgetattr, tcgetwinsize, tcsetattr}; use rustix::termios::{LocalModes, OptionalActions, Termios, tcgetattr, tcgetwinsize, tcsetattr};
use signal_hook::SigId;
use signal_hook::consts::signal::{SIGHUP, SIGINT, SIGQUIT, SIGTERM, SIGTSTP, SIGWINCH};
use crate::{ClientError, ClientSessionIdentity}; use crate::{ClientError, ClientSessionIdentity};
@ -64,18 +62,24 @@ pub trait HostTerminal {
fn write_output(&mut self, bytes: &[u8]) -> io::Result<()>; fn write_output(&mut self, bytes: &[u8]) -> io::Result<()>;
/// # Errors /// # Errors
/// Returns an error while collecting one explicit developer decision. /// Returns an error while collecting one explicit developer decision.
fn prompt(&mut self, prompt: &TerminalPrompt) -> io::Result<DeveloperDecisionKind>; fn prompt(&mut self, prompt: &TerminalPrompt) -> io::Result<PromptOutcome>;
/// # Errors /// # Errors
/// Returns an error restoring, stopping, or re-entering raw mode. /// Returns an error restoring, stopping, or re-entering raw mode.
fn suspend_and_resume(&mut self) -> io::Result<()>; fn suspend_and_resume(&mut self) -> io::Result<()>;
} }
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PromptOutcome {
pub decision: DeveloperDecisionKind,
pub resumed_size: Option<TerminalResize>,
}
pub struct SystemTerminal { pub struct SystemTerminal {
terminal: File, terminal: File,
original: Termios, original: Termios,
original_flags: OFlags, original_flags: OFlags,
raw: bool, raw: bool,
signals: SignalFlags, signals: SignalQueue,
} }
impl SystemTerminal { impl SystemTerminal {
@ -90,7 +94,7 @@ impl SystemTerminal {
original_flags: fcntl_getfl(&terminal)?, original_flags: fcntl_getfl(&terminal)?,
terminal, terminal,
raw: false, raw: false,
signals: SignalFlags::register()?, signals: SignalQueue::register()?,
}) })
} }
} }
@ -136,20 +140,8 @@ impl HostTerminal for SystemTerminal {
} }
fn next_event(&mut self) -> io::Result<HostTerminalEvent> { fn next_event(&mut self) -> io::Result<HostTerminalEvent> {
if self.signals.terminate.swap(false, Ordering::SeqCst) { if let Some(event) = self.signals.next_event()? {
return Ok(HostTerminalEvent::Terminate); return Ok(event);
}
if self.signals.suspend.swap(false, Ordering::SeqCst) {
return Ok(HostTerminalEvent::LocalSuspend);
}
if self.signals.resize.swap(false, Ordering::SeqCst) {
return Ok(HostTerminalEvent::Resize);
}
if self.signals.interrupt.swap(false, Ordering::SeqCst) {
return Ok(HostTerminalEvent::Signal(TerminalSignalKind::Interrupt));
}
if self.signals.quit.swap(false, Ordering::SeqCst) {
return Ok(HostTerminalEvent::Signal(TerminalSignalKind::Quit));
} }
let mut bytes = vec![0_u8; 16 * 1024]; let mut bytes = vec![0_u8; 16 * 1024];
match self.terminal.read(&mut bytes) { match self.terminal.read(&mut bytes) {
@ -171,7 +163,7 @@ impl HostTerminal for SystemTerminal {
self.terminal.flush() self.terminal.flush()
} }
fn prompt(&mut self, prompt: &TerminalPrompt) -> io::Result<DeveloperDecisionKind> { fn prompt(&mut self, prompt: &TerminalPrompt) -> io::Result<PromptOutcome> {
self.write_output( self.write_output(
format!( format!(
"\r\nLazaret host request\r\nprofile: {}\r\nsession: {}\r\ncommand: {}\r\njustification: {}\r\nApprove? [y/N] ", "\r\nLazaret host request\r\nprofile: {}\r\nsession: {}\r\ncommand: {}\r\njustification: {}\r\nApprove? [y/N] ",
@ -182,6 +174,7 @@ impl HostTerminal for SystemTerminal {
) )
.as_bytes(), .as_bytes(),
)?; )?;
let mut resumed_size = None;
loop { loop {
match self.next_event()? { match self.next_event()? {
HostTerminalEvent::Input(bytes) => { HostTerminalEvent::Input(bytes) => {
@ -189,13 +182,19 @@ impl HostTerminal for SystemTerminal {
.first() .first()
.is_some_and(|byte| matches!(byte, b'y' | b'Y')); .is_some_and(|byte| matches!(byte, b'y' | b'Y'));
self.write_output(if approve { b"y\r\n" } else { b"n\r\n" })?; self.write_output(if approve { b"y\r\n" } else { b"n\r\n" })?;
return Ok(if approve { return Ok(PromptOutcome {
decision: if approve {
DeveloperDecisionKind::Approve DeveloperDecisionKind::Approve
} else { } else {
DeveloperDecisionKind::Deny DeveloperDecisionKind::Deny
},
resumed_size,
}); });
} }
HostTerminalEvent::LocalSuspend => self.suspend_and_resume()?, HostTerminalEvent::LocalSuspend => {
self.suspend_and_resume()?;
resumed_size = Some(self.size(prompt.session_id)?);
}
HostTerminalEvent::Terminate => return Err(io::ErrorKind::Interrupted.into()), HostTerminalEvent::Terminate => return Err(io::ErrorKind::Interrupted.into()),
HostTerminalEvent::Idle HostTerminalEvent::Idle
| HostTerminalEvent::Resize | HostTerminalEvent::Resize
@ -214,52 +213,79 @@ impl HostTerminal for SystemTerminal {
impl Drop for SystemTerminal { impl Drop for SystemTerminal {
fn drop(&mut self) { fn drop(&mut self) {
let _ = self.restore(); let _ = self.restore();
self.signals.restore();
} }
} }
struct SignalFlags { struct SignalQueue {
interrupt: Arc<AtomicBool>, fd: SignalFd,
quit: Arc<AtomicBool>, original_mask: SigSet,
suspend: Arc<AtomicBool>, restored: bool,
resize: Arc<AtomicBool>,
terminate: Arc<AtomicBool>,
registrations: Vec<SigId>,
} }
impl SignalFlags { impl SignalQueue {
fn register() -> io::Result<Self> { fn register() -> io::Result<Self> {
let interrupt = Arc::new(AtomicBool::new(false)); let mut mask = SigSet::empty();
let quit = Arc::new(AtomicBool::new(false)); for signal in [
let suspend = Arc::new(AtomicBool::new(false)); Signal::SIGINT,
let resize = Arc::new(AtomicBool::new(false)); Signal::SIGQUIT,
let terminate = Arc::new(AtomicBool::new(false)); Signal::SIGTSTP,
let registrations = vec![ Signal::SIGWINCH,
signal_hook::flag::register(SIGINT, Arc::clone(&interrupt))?, Signal::SIGTERM,
signal_hook::flag::register(SIGQUIT, Arc::clone(&quit))?, Signal::SIGHUP,
signal_hook::flag::register(SIGTSTP, Arc::clone(&suspend))?, ] {
signal_hook::flag::register(SIGWINCH, Arc::clone(&resize))?, mask.add(signal);
signal_hook::flag::register(SIGTERM, Arc::clone(&terminate))?, }
signal_hook::flag::register(SIGHUP, Arc::clone(&terminate))?, let original_mask = mask
]; .thread_swap_mask(SigmaskHow::SIG_BLOCK)
.map_err(errno_to_io)?;
let fd = match SignalFd::with_flags(&mask, SfdFlags::SFD_NONBLOCK | SfdFlags::SFD_CLOEXEC) {
Ok(fd) => fd,
Err(error) => {
let _ = original_mask.thread_swap_mask(SigmaskHow::SIG_SETMASK);
return Err(errno_to_io(error));
}
};
Ok(Self { Ok(Self {
interrupt, fd,
quit, original_mask,
suspend, restored: false,
resize,
terminate,
registrations,
}) })
} }
fn next_event(&mut self) -> io::Result<Option<HostTerminalEvent>> {
let Some(info) = self.fd.read_signal().map_err(errno_to_io)? else {
return Ok(None);
};
let signal = Signal::try_from(info.ssi_signo.cast_signed()).map_err(errno_to_io)?;
Ok(match signal {
Signal::SIGINT => Some(HostTerminalEvent::Signal(TerminalSignalKind::Interrupt)),
Signal::SIGQUIT => Some(HostTerminalEvent::Signal(TerminalSignalKind::Quit)),
Signal::SIGTSTP => Some(HostTerminalEvent::LocalSuspend),
Signal::SIGWINCH => Some(HostTerminalEvent::Resize),
Signal::SIGTERM | Signal::SIGHUP => Some(HostTerminalEvent::Terminate),
_ => None,
})
}
fn restore(&mut self) {
if !self.restored {
let _ = self.original_mask.thread_swap_mask(SigmaskHow::SIG_SETMASK);
self.restored = true;
}
}
} }
impl Drop for SignalFlags { impl Drop for SignalQueue {
fn drop(&mut self) { fn drop(&mut self) {
for registration in self.registrations.drain(..) { self.restore();
signal_hook::low_level::unregister(registration);
}
} }
} }
fn errno_to_io(error: nix::errno::Errno) -> io::Error {
io::Error::from_raw_os_error(error as i32)
}
/// Runs one interactive terminal route until its exact Session exit. /// Runs one interactive terminal route until its exact Session exit.
/// ///
/// # Errors /// # Errors
@ -344,9 +370,19 @@ fn handle_batch<T: InteractiveSessionTransport, H: HostTerminal>(
} }
InteractiveSessionMessage::Prompt(prompt) => { InteractiveSessionMessage::Prompt(prompt) => {
require_session(prompt.session_id, identity.session_id())?; require_session(prompt.session_id, identity.session_id())?;
let decision = terminal.prompt(&prompt)?; let outcome = terminal.prompt(&prompt)?;
if let Some(resize) = outcome.resumed_size {
let response = transport.exchange(InteractiveSessionEvent::Terminal(
SessionTerminalEvent::Resize(resize),
))?;
if let Some(exit) =
handle_batch(transport, terminal, identity, binding, response)?
{
return Ok(Some(exit));
}
}
let signed_decision = let signed_decision =
identity.sign_developer_decision(binding.clone(), &prompt, decision)?; identity.sign_developer_decision(binding.clone(), &prompt, outcome.decision)?;
let response = transport.exchange(InteractiveSessionEvent::PromptDecision { let response = transport.exchange(InteractiveSessionEvent::PromptDecision {
session_id: identity.session_id(), session_id: identity.session_id(),
signed_decision, signed_decision,
@ -387,3 +423,44 @@ fn hex(bytes: &[u8]) -> String {
output output
}) })
} }
#[cfg(test)]
#[allow(clippy::disallowed_methods, clippy::disallowed_types)]
mod tests {
use std::os::unix::process::ExitStatusExt as _;
use std::process::Command;
use nix::sys::signal::{SigSet, SigmaskHow, Signal, kill};
use nix::unistd::Pid;
use super::SignalQueue;
#[test]
fn signal_scope_restores_mask_and_default_termination() {
if std::env::var_os("LAZARET_SIGNAL_SCOPE_CHILD").is_some() {
let signals = SignalQueue::register().unwrap();
drop(signals);
kill(Pid::this(), Signal::SIGTERM).unwrap();
panic!("restored SIGTERM did not terminate the process");
}
let observed = SigSet::empty()
.thread_swap_mask(SigmaskHow::SIG_BLOCK)
.unwrap();
let signals = SignalQueue::register().unwrap();
drop(signals);
let restored = SigSet::empty()
.thread_swap_mask(SigmaskHow::SIG_BLOCK)
.unwrap();
assert_eq!(restored, observed);
let status = Command::new(std::env::current_exe().unwrap())
.arg("--exact")
.arg("terminal::tests::signal_scope_restores_mask_and_default_termination")
.arg("--nocapture")
.env("LAZARET_SIGNAL_SCOPE_CHILD", "1")
.status()
.unwrap();
assert_eq!(status.signal(), Some(nix::libc::SIGTERM));
}
}

View file

@ -3,7 +3,8 @@ use std::fs::File;
use std::io; use std::io;
use lazaret_client::terminal::{ use lazaret_client::terminal::{
HostTerminal, HostTerminalEvent, InteractiveSessionTransport, run_interactive_session, HostTerminal, HostTerminalEvent, InteractiveSessionTransport, PromptOutcome,
run_interactive_session,
}; };
use lazaret_client::{ClientError, ClientSessionIdentity}; use lazaret_client::{ClientError, ClientSessionIdentity};
use lazaret_core::identity::{ use lazaret_core::identity::{
@ -106,11 +107,21 @@ impl HostTerminal for PtyTerminal {
Ok(()) Ok(())
} }
fn prompt(&mut self, prompt: &TerminalPrompt) -> io::Result<DeveloperDecisionKind> { fn prompt(&mut self, prompt: &TerminalPrompt) -> io::Result<PromptOutcome> {
assert_eq!(prompt.agent_profile, "trusted"); assert_eq!(prompt.agent_profile, "trusted");
assert_eq!(prompt.command_argv, [b"/usr/bin/tool".to_vec()]); assert_eq!(prompt.command_argv, [b"/usr/bin/tool".to_vec()]);
assert_eq!(prompt.justification, "update host metadata"); assert_eq!(prompt.justification, "update host metadata");
Ok(self.decision) let resumed_size = if matches!(self.events.front(), Some(HostTerminalEvent::LocalSuspend)) {
self.events.pop_front();
self.suspend_and_resume()?;
Some(self.size(prompt.session_id)?)
} else {
None
};
Ok(PromptOutcome {
decision: self.decision,
resumed_size,
})
} }
fn suspend_and_resume(&mut self) -> io::Result<()> { fn suspend_and_resume(&mut self) -> io::Result<()> {
@ -132,6 +143,7 @@ struct FixtureTransport {
prompt_sent: bool, prompt_sent: bool,
failure: Option<TransportFailure>, failure: Option<TransportFailure>,
signed_decision: Option<Vec<u8>>, signed_decision: Option<Vec<u8>>,
prompt_after_events: usize,
} }
impl FixtureTransport { impl FixtureTransport {
@ -142,6 +154,7 @@ impl FixtureTransport {
prompt_sent: false, prompt_sent: false,
failure: None, failure: None,
signed_decision: None, signed_decision: None,
prompt_after_events: 5,
} }
} }
} }
@ -188,7 +201,7 @@ impl InteractiveSessionTransport for FixtureTransport {
)], )],
)); ));
} }
if !self.prompt_sent && self.events.len() >= 5 { if !self.prompt_sent && self.events.len() >= self.prompt_after_events {
self.prompt_sent = true; self.prompt_sent = true;
return Ok(batch( return Ok(batch(
self.session_id, self.session_id,
@ -206,6 +219,39 @@ impl InteractiveSessionTransport for FixtureTransport {
} }
} }
#[test]
fn prompt_suspend_sends_resumed_size_before_decision() {
let identity = ClientSessionIdentity::generate::<std::convert::Infallible>().unwrap();
let binding = binding(&identity);
let mut terminal = PtyTerminal::new([HostTerminalEvent::Idle, HostTerminalEvent::LocalSuspend]);
let mut transport = FixtureTransport::successful(identity.session_id());
transport.prompt_after_events = 1;
assert_eq!(
run_interactive_session(
&mut transport,
&mut terminal,
&identity,
&binding,
&[b"/bin/sh".to_vec()],
)
.unwrap(),
23
);
assert_eq!(terminal.suspend_count, 1);
assert!(matches!(
transport.events.as_slice(),
[
InteractiveSessionEvent::Poll { .. },
InteractiveSessionEvent::Terminal(lazaret_core::records::SessionTerminalEvent::Resize(
_
)),
InteractiveSessionEvent::PromptDecision { .. }
]
));
}
#[test] #[test]
fn terminal_fixture_routes_resize_signals_suspend_prompt_and_nonzero_exit() { fn terminal_fixture_routes_resize_signals_suspend_prompt_and_nonzero_exit() {
let identity = ClientSessionIdentity::generate::<std::convert::Infallible>().unwrap(); let identity = ClientSessionIdentity::generate::<std::convert::Infallible>().unwrap();

View file

@ -34,6 +34,7 @@ use crate::runtime::{
StateAnchorAbsenceObservation, StateAnchorAbsenceObservation,
}; };
use crate::state_anchor::{StateAnchorCoordinator, StateAnchorReleaseAuthority}; use crate::state_anchor::{StateAnchorCoordinator, StateAnchorReleaseAuthority};
use crate::terminal::SessionPromptTransport;
#[derive(Debug)] #[derive(Debug)]
pub struct ProductionLifecycleError(String); pub struct ProductionLifecycleError(String);
@ -74,6 +75,7 @@ pub struct ProductionBootstrapPlan<'a> {
pub invoking_uid: u32, pub invoking_uid: u32,
pub non_interactive_executor: Arc<dyn NonInteractiveSessionExecutor>, pub non_interactive_executor: Arc<dyn NonInteractiveSessionExecutor>,
pub interactive_executor: Arc<dyn InteractiveSessionExecutor>, pub interactive_executor: Arc<dyn InteractiveSessionExecutor>,
pub prompt_transport: Arc<SessionPromptTransport>,
pub(crate) runtime_staging: &'a crate::production::RuntimeStaging, pub(crate) runtime_staging: &'a crate::production::RuntimeStaging,
} }

View file

@ -325,6 +325,16 @@ impl<'a> ProductionColdStart<'a> {
} }
fn take_plan(&mut self) -> Result<ProductionBootstrapPlan<'_>, ProductionColdStartError> { fn take_plan(&mut self) -> Result<ProductionBootstrapPlan<'_>, ProductionColdStartError> {
let interactive_executor = RunscInteractiveSessionExecutor::new(
Arc::clone(&self.store),
&self.generation_identity,
self.container_id,
current_process_identity().map_err(failure)?,
self.sandbox_bundle.command(),
self.runtime_root.clone(),
)
.map_err(failure)?;
let prompt_transport = interactive_executor.prompt_transport();
Ok(ProductionBootstrapPlan { Ok(ProductionBootstrapPlan {
store: Arc::clone(&self.store), store: Arc::clone(&self.store),
catalog: self.catalog, catalog: self.catalog,
@ -350,17 +360,8 @@ impl<'a> ProductionColdStart<'a> {
) )
.map_err(failure)?, .map_err(failure)?,
), ),
interactive_executor: Arc::new( interactive_executor: Arc::new(interactive_executor),
RunscInteractiveSessionExecutor::new( prompt_transport,
Arc::clone(&self.store),
&self.generation_identity,
self.container_id,
current_process_identity().map_err(failure)?,
self.sandbox_bundle.command(),
self.runtime_root.clone(),
)
.map_err(failure)?,
),
runtime_staging: &self.staging, runtime_staging: &self.staging,
}) })
} }

View file

@ -13,13 +13,14 @@ use lazaret_core::identity::{
GuestProcessGroupId, GuestProcessId, GuestProcessStartTimeTicks, InodeId, GuestProcessGroupId, GuestProcessId, GuestProcessStartTimeTicks, InodeId,
PidfsProcessIdentityV1, SessionId, PidfsProcessIdentityV1, SessionId,
}; };
use lazaret_core::protocol::VerifiedSession;
use lazaret_core::records::{ use lazaret_core::records::{
DomainStore, DurableDecision, GuestProcessIdentity, InteractiveSessionBatch, DeveloperDecision, DomainStore, DurableDecision, GuestProcessIdentity, InteractiveSessionBatch,
InteractiveSessionEvent, InteractiveSessionMessage, InteractiveSessionRequest, InteractiveSessionEvent, InteractiveSessionMessage, InteractiveSessionRequest,
ParticipantOwnership, ParticipantRole, ParticipantState, PersistedHostProcessOwnership, ParticipantOwnership, ParticipantRole, ParticipantState, PersistedHostProcessOwnership,
ProcessOwnershipCommitAck, ProtocolVersion, PtyIdentity, SessionExit, SessionPhase, ProcessOwnershipCommitAck, ProtocolVersion, PtyIdentity, SessionExit, SessionPhase,
SessionTerminal, SessionTerminalEvent, SessionTerminalFields, SessionTerminalMessage, SessionTerminal, SessionTerminalEvent, SessionTerminalFields, SessionTerminalMessage,
SessionTerminalState, StagedProcessOwnership, TerminalOutput, TerminalResize, SessionTerminalState, StagedProcessOwnership, TerminalOutput, TerminalPrompt, TerminalResize,
TerminalSignalKind, TerminalSignalKind,
}; };
use lazaret_linux::process::{ use lazaret_linux::process::{
@ -29,6 +30,7 @@ use lazaret_linux::process::{
use rustix::fs::{OFlags, fcntl_getfl, fcntl_setfl, fstat}; use rustix::fs::{OFlags, fcntl_getfl, fcntl_setfl, fstat};
use rustix::pty::{OpenptFlags, grantpt, ioctl_tiocgptpeer, openpt, unlockpt}; use rustix::pty::{OpenptFlags, grantpt, ioctl_tiocgptpeer, openpt, unlockpt};
use rustix::termios::{Winsize, tcsetwinsize}; use rustix::termios::{Winsize, tcsetwinsize};
use sha2::{Digest, Sha256};
use crate::control::{ControlError, InteractiveControlSession, InteractiveSessionExecutor}; use crate::control::{ControlError, InteractiveControlSession, InteractiveSessionExecutor};
use crate::domain_store::{DomainStoreError, DomainStoreMutation, HostCoordinatorDomainStore}; use crate::domain_store::{DomainStoreError, DomainStoreMutation, HostCoordinatorDomainStore};
@ -87,6 +89,8 @@ pub enum TerminalError {
GuestGroupStillPresent, GuestGroupStillPresent,
InvalidExit, InvalidExit,
AuthorityMismatch, AuthorityMismatch,
DuplicatePrompt,
PromptValidation,
AcquisitionCleanup { AcquisitionCleanup {
original: Box<TerminalError>, original: Box<TerminalError>,
cleanup: Option<Box<TerminalError>>, cleanup: Option<Box<TerminalError>>,
@ -119,6 +123,8 @@ impl std::fmt::Display for TerminalError {
Self::AuthorityMismatch => { Self::AuthorityMismatch => {
formatter.write_str("Session Terminal runtime authority does not match") formatter.write_str("Session Terminal runtime authority does not match")
} }
Self::DuplicatePrompt => formatter.write_str("Session already has an active prompt"),
Self::PromptValidation => formatter.write_str("prompt authority validation failed"),
Self::AcquisitionCleanup { original, cleanup } => { Self::AcquisitionCleanup { original, cleanup } => {
write!(formatter, "Session Terminal acquisition failed: {original}")?; write!(formatter, "Session Terminal acquisition failed: {original}")?;
if let Some(cleanup) = cleanup { if let Some(cleanup) = cleanup {
@ -146,7 +152,9 @@ impl std::error::Error for TerminalError {
| Self::GuestIdentityMismatch | Self::GuestIdentityMismatch
| Self::GuestGroupStillPresent | Self::GuestGroupStillPresent
| Self::InvalidExit | Self::InvalidExit
| Self::AuthorityMismatch => None, | Self::AuthorityMismatch
| Self::DuplicatePrompt
| Self::PromptValidation => None,
} }
} }
} }
@ -176,6 +184,185 @@ impl PromptRouteCancellation for NoActivePrompt {
} }
} }
struct PendingPrompt {
authority: VerifiedSession,
prompt: TerminalPrompt,
emitted: bool,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RoutedPromptDecision {
pub signed_decision: Vec<u8>,
pub decision: DeveloperDecision,
}
pub struct SessionPromptTransport {
store: Arc<HostCoordinatorDomainStore>,
pending: Mutex<HashMap<SessionId, PendingPrompt>>,
routed: Mutex<HashMap<SessionId, RoutedPromptDecision>>,
}
impl SessionPromptTransport {
#[must_use]
pub fn new(store: Arc<HostCoordinatorDomainStore>) -> Self {
Self {
store,
pending: Mutex::new(HashMap::new()),
routed: Mutex::new(HashMap::new()),
}
}
/// Queues one sandbox-signed request for its exact registered Session.
///
/// # Errors
/// Refuses invalid authority, stale Session state, and concurrent prompts.
pub fn queue_signed_request(
&self,
authority: VerifiedSession,
signed_request: &[u8],
) -> Result<TerminalPrompt, TerminalError> {
let (request, digest) = authority
.verify_request(signed_request)
.map_err(|_| TerminalError::PromptValidation)?;
let binding = authority.session_binding();
let anchor_id = lazaret_core::identity::SessionAnchorId::new(
Sha256::digest(&binding.session_anchor_public_key).into(),
);
let snapshot = self.store.snapshot()?;
let mut session = snapshot
.sessions
.into_iter()
.find(|session| {
session.generation_id == binding.generation_identity.generation_id
&& session.session_id == binding.session_id
&& session.cli_id == binding.cli_id
&& session.session_anchor_id == anchor_id
&& session.session_phase == SessionPhase::Attached
})
.ok_or(TerminalError::AuthorityMismatch)?;
let prompt = TerminalPrompt {
session_id: binding.session_id,
request_id: request.request_id(),
request_sha256: digest,
agent_profile: binding.generation_identity.agent_profile.clone(),
command_argv: request.command_argv().to_vec(),
justification: request.justification().to_owned(),
};
let mut pending = self
.pending
.lock()
.map_err(|_| TerminalError::PromptValidation)?;
if pending.contains_key(&binding.session_id) {
return Err(TerminalError::DuplicatePrompt);
}
session.session_phase = SessionPhase::Prompting;
self.store
.apply(&[DomainStoreMutation::PutSession(session)])?;
pending.insert(
binding.session_id,
PendingPrompt {
authority,
prompt: prompt.clone(),
emitted: false,
},
);
Ok(prompt)
}
fn prompt_for(&self, session_id: SessionId) -> Result<Option<TerminalPrompt>, TerminalError> {
let mut pending = self
.pending
.lock()
.map_err(|_| TerminalError::PromptValidation)?;
let Some(pending) = pending.get_mut(&session_id) else {
return Ok(None);
};
if pending.emitted {
return Ok(None);
}
pending.emitted = true;
Ok(Some(pending.prompt.clone()))
}
fn route_decision(
&self,
session_id: SessionId,
signed_decision: Vec<u8>,
) -> Result<DeveloperDecision, TerminalError> {
let mut prompts = self
.pending
.lock()
.map_err(|_| TerminalError::PromptValidation)?;
let pending = prompts
.get(&session_id)
.ok_or(TerminalError::InactiveRoute)?;
let decision = pending
.authority
.verify_developer_decision(
&signed_decision,
pending.prompt.request_id,
pending.prompt.request_sha256,
)
.map_err(|_| TerminalError::PromptValidation)?;
prompts.remove(&session_id);
drop(prompts);
self.routed
.lock()
.map_err(|_| TerminalError::PromptValidation)?
.insert(
session_id,
RoutedPromptDecision {
signed_decision,
decision: decision.clone(),
},
);
Ok(decision)
}
fn restore_sandbox_route(&self, session_id: SessionId) -> 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)?;
if session.session_phase != SessionPhase::Prompting {
return Err(TerminalError::InactiveRoute);
}
session.session_phase = SessionPhase::Attached;
self.store
.apply(&[DomainStoreMutation::PutSession(session)])?;
Ok(())
}
/// # Errors
/// Refuses access after internal synchronization failure.
pub fn take_decision(
&self,
session_id: SessionId,
) -> Result<Option<RoutedPromptDecision>, TerminalError> {
Ok(self
.routed
.lock()
.map_err(|_| TerminalError::PromptValidation)?
.remove(&session_id))
}
}
impl PromptRouteCancellation for SessionPromptTransport {
fn cancel_for_close(&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);
Ok(())
}
}
#[allow(clippy::missing_errors_doc)] #[allow(clippy::missing_errors_doc)]
pub trait SessionTerminalProcess: Send { pub trait SessionTerminalProcess: Send {
fn write_input(&mut self, bytes: &[u8]) -> io::Result<usize>; fn write_input(&mut self, bytes: &[u8]) -> io::Result<usize>;
@ -411,6 +598,7 @@ pub struct SessionTerminalEngine {
pub struct RunscInteractiveSessionExecutor { pub struct RunscInteractiveSessionExecutor {
engine: Arc<Mutex<SessionTerminalEngine>>, engine: Arc<Mutex<SessionTerminalEngine>>,
prompt_transport: Arc<SessionPromptTransport>,
domain_id: DomainId, domain_id: DomainId,
generation_id: GenerationId, generation_id: GenerationId,
coordinator_id: CoordinatorId, coordinator_id: CoordinatorId,
@ -442,12 +630,14 @@ impl RunscInteractiveSessionExecutor {
if run + 3 != command.arguments().len() { if run + 3 != command.arguments().len() {
return Err(TerminalError::AuthorityMismatch); return Err(TerminalError::AuthorityMismatch);
} }
let prompt_transport = Arc::new(SessionPromptTransport::new(Arc::clone(&store)));
Ok(Self { Ok(Self {
engine: Arc::new(Mutex::new(SessionTerminalEngine::new( engine: Arc::new(Mutex::new(SessionTerminalEngine::new(
store, store,
Arc::new(RunscSessionTerminalFactory), Arc::new(RunscSessionTerminalFactory),
Arc::new(NoActivePrompt), prompt_transport.clone(),
))), ))),
prompt_transport,
domain_id: generation.domain_id, domain_id: generation.domain_id,
generation_id: generation.generation_id, generation_id: generation.generation_id,
coordinator_id: generation.coordinator_id, coordinator_id: generation.coordinator_id,
@ -458,6 +648,11 @@ impl RunscInteractiveSessionExecutor {
runtime_root, runtime_root,
}) })
} }
#[must_use]
pub fn prompt_transport(&self) -> Arc<SessionPromptTransport> {
Arc::clone(&self.prompt_transport)
}
} }
impl InteractiveSessionExecutor for RunscInteractiveSessionExecutor { impl InteractiveSessionExecutor for RunscInteractiveSessionExecutor {
@ -503,6 +698,7 @@ impl InteractiveSessionExecutor for RunscInteractiveSessionExecutor {
drop(engine); drop(engine);
Ok(Box::new(RunscInteractiveControlSession { Ok(Box::new(RunscInteractiveControlSession {
engine: Arc::clone(&self.engine), engine: Arc::clone(&self.engine),
prompt_transport: Arc::clone(&self.prompt_transport),
generation_id: self.generation_id, generation_id: self.generation_id,
session_id: request.session_id, session_id: request.session_id,
})) }))
@ -511,6 +707,7 @@ impl InteractiveSessionExecutor for RunscInteractiveSessionExecutor {
struct RunscInteractiveControlSession { struct RunscInteractiveControlSession {
engine: Arc<Mutex<SessionTerminalEngine>>, engine: Arc<Mutex<SessionTerminalEngine>>,
prompt_transport: Arc<SessionPromptTransport>,
generation_id: GenerationId, generation_id: GenerationId,
session_id: SessionId, session_id: SessionId,
} }
@ -528,9 +725,40 @@ impl InteractiveControlSession for RunscInteractiveControlSession {
InteractiveSessionEvent::Poll { session_id } if session_id == self.session_id => { InteractiveSessionEvent::Poll { session_id } if session_id == self.session_id => {
Vec::new() Vec::new()
} }
InteractiveSessionEvent::PromptDecision {
session_id,
signed_decision,
} if session_id == self.session_id => {
let decision = self
.prompt_transport
.route_decision(session_id, signed_decision)
.map_err(|error| terminal_control_error(&error))?;
if decision.decision == lazaret_core::records::DeveloperDecisionKind::Deny {
self.prompt_transport
.restore_sandbox_route(session_id)
.map_err(|error| terminal_control_error(&error))?;
engine
.set_route(session_id, self.generation_id, TerminalRoute::Sandbox)
.map_err(|error| terminal_control_error(&error))?;
}
Vec::new()
}
InteractiveSessionEvent::PromptDecision { .. } InteractiveSessionEvent::PromptDecision { .. }
| InteractiveSessionEvent::Poll { .. } => return Err(ControlError::InvalidRequest), | InteractiveSessionEvent::Poll { .. } => return Err(ControlError::InvalidRequest),
}; };
if let Some(prompt) = self
.prompt_transport
.prompt_for(self.session_id)
.map_err(|error| terminal_control_error(&error))?
{
engine
.set_route(self.session_id, self.generation_id, TerminalRoute::Prompt)
.map_err(|error| terminal_control_error(&error))?;
return Ok(InteractiveSessionBatch {
session_id: self.session_id,
messages: vec![InteractiveSessionMessage::Prompt(prompt)],
});
}
if !messages if !messages
.iter() .iter()
.any(|message| matches!(message, SessionTerminalMessage::Exit(_))) .any(|message| matches!(message, SessionTerminalMessage::Exit(_)))

View file

@ -4,18 +4,22 @@ use std::io;
use std::num::NonZeroU16; use std::num::NonZeroU16;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use ed25519_dalek::SigningKey;
use lazaret_core::identity::{ use lazaret_core::identity::{
BootId, BundleDigest, ClientId, ConfigurationFingerprint, HostPath, HostPidLocator, BootId, BundleDigest, ClientId, ConfigurationFingerprint, HostPath, HostPidLocator,
NetworkSidecarId, PidfsDeviceId, PidfsInodeId, PidfsProcessAuthorityV1, ProcessStartTimeTicks, NetworkSidecarId, PidfsDeviceId, PidfsInodeId, PidfsProcessAuthorityV1, ProcessStartTimeTicks,
SandboxId, SessionAnchorId, RequestId, SandboxId, SessionAnchorId,
}; };
use lazaret_core::protocol::{VerifiedSession, encode_signed, issue_session_capability};
use lazaret_core::records::{ use lazaret_core::records::{
Domain, DurableDecision, Generation, Participant, SandboxOwnership, Session, TerminalInput, DeveloperDecisionKind, Domain, DurableDecision, Generation, GenerationIdentity, MailboxRequest,
TerminalSignal, Participant, SandboxOwnership, Session, SessionBinding, TerminalInput, TerminalSignal,
}; };
use lazaret_linux::filesystem::try_exclusive_flock; use lazaret_linux::filesystem::try_exclusive_flock;
use tempfile::TempDir; use tempfile::TempDir;
use crate::control::TrustAnchor;
use super::*; use super::*;
fn bytes(value: u8) -> [u8; 32] { fn bytes(value: u8) -> [u8; 32] {
@ -533,6 +537,158 @@ fn engine_for(fixture: &Fixture) -> (Arc<MockFactory>, SessionTerminalEngine) {
engine_for_store(store) engine_for_store(store)
} }
#[test]
#[allow(clippy::too_many_lines)]
fn production_prompt_adapter_routes_exact_authenticated_denial() {
let generation = GenerationIdentity {
agent_profile: "trusted".into(),
domain_id: DomainId::new(bytes(1)),
generation_id: GenerationId::new(7),
configuration_fingerprint: ConfigurationFingerprint::new(bytes(2)),
coordinator_id: CoordinatorId::new(bytes(3)),
sandbox_id: SandboxId::new(bytes(5)),
};
let mut trust_anchor = TrustAnchor::create(generation.clone()).unwrap();
let publication = trust_anchor.publication_identity();
let coordinator_key = trust_anchor.coordinator_protocol_key();
let credentials = trust_anchor.take_sandbox_credential_bundle().unwrap();
let sandbox_key = SigningKey::from_bytes(
credentials
.sandbox_private_key
.as_slice()
.try_into()
.unwrap(),
);
let anchor_key = SigningKey::from_bytes(&bytes(61));
let binding = SessionBinding {
generation_identity: generation,
cli_id: ClientId::new(bytes(62)),
session_id: SessionId::new(bytes(63)),
session_anchor_public_key: anchor_key.verifying_key().to_bytes().to_vec(),
};
let capability = issue_session_capability(
binding.clone(),
credentials.coordinator_protocol_credential().credential_id,
credentials.sandbox_protocol_credential().credential_id,
&coordinator_key,
)
.unwrap();
let authority = VerifiedSession::new(
publication.trust_manifest_sha256,
credentials.trust_manifest.tls_ca_sha256,
&credentials.trust_manifest,
binding.clone(),
credentials.coordinator_protocol_credential(),
credentials.sandbox_protocol_credential(),
capability.clone(),
)
.unwrap();
let fixture = Fixture::new(&[]);
let session_anchor_id =
SessionAnchorId::new(Sha256::digest(&binding.session_anchor_public_key).into());
let mut session = Session {
generation_id: GenerationId::new(7),
session_id: binding.session_id,
cli_id: binding.cli_id,
session_anchor_id,
session_phase: SessionPhase::Registering,
};
fixture
.store
.apply(&[DomainStoreMutation::PutSession(session.clone())])
.unwrap();
session.session_phase = SessionPhase::Attached;
fixture
.store
.apply(&[DomainStoreMutation::PutSession(session)])
.unwrap();
let prompt_transport = Arc::new(SessionPromptTransport::new(Arc::clone(&fixture.store)));
let factory = Arc::new(MockFactory::default());
let mut engine = SessionTerminalEngine::new(
Arc::clone(&fixture.store),
factory,
prompt_transport.clone(),
);
engine.acquire(&acquisition(binding.session_id)).unwrap();
let request = MailboxRequest::new(
binding.clone(),
RequestId::new(bytes(64)),
capability,
credentials.sandbox_protocol_credential().credential_id,
vec![b"/usr/bin/tool".to_vec()],
"update host metadata".into(),
)
.unwrap();
let signed_request = encode_signed(&request, &sandbox_key).unwrap();
let expected_prompt = prompt_transport
.queue_signed_request(authority, &signed_request)
.unwrap();
let engine = Arc::new(Mutex::new(engine));
let mut control = RunscInteractiveControlSession {
engine: Arc::clone(&engine),
prompt_transport: prompt_transport.clone(),
generation_id: GenerationId::new(7),
session_id: binding.session_id,
};
let batch = control
.exchange(InteractiveSessionEvent::Poll {
session_id: binding.session_id,
})
.unwrap();
assert_eq!(
batch.messages,
vec![InteractiveSessionMessage::Prompt(expected_prompt.clone())]
);
let signed_decision = encode_signed(
&DeveloperDecision {
protocol_version: ProtocolVersion::V1,
session_binding: binding.clone(),
request_id: expected_prompt.request_id,
request_sha256: expected_prompt.request_sha256,
decision: DeveloperDecisionKind::Deny,
approved_command_argv: Vec::new(),
},
&anchor_key,
)
.unwrap();
control
.exchange(InteractiveSessionEvent::PromptDecision {
session_id: binding.session_id,
signed_decision: signed_decision.clone(),
})
.unwrap();
let routed = prompt_transport
.take_decision(binding.session_id)
.unwrap()
.unwrap();
assert_eq!(routed.signed_decision, signed_decision);
assert_eq!(routed.decision.decision, DeveloperDecisionKind::Deny);
assert_eq!(
fixture
.store
.snapshot()
.unwrap()
.sessions
.into_iter()
.find(|session| session.session_id == binding.session_id)
.unwrap()
.session_phase,
SessionPhase::Attached
);
engine
.lock()
.unwrap()
.handle(
GenerationId::new(7),
SessionTerminalEvent::Input(TerminalInput {
session_id: binding.session_id,
bytes: b"sandbox-input".to_vec(),
}),
)
.unwrap();
}
fn engine_for_store( fn engine_for_store(
store: Arc<dyn TerminalDomainStore>, store: Arc<dyn TerminalDomainStore>,
) -> (Arc<MockFactory>, SessionTerminalEngine) { ) -> (Arc<MockFactory>, SessionTerminalEngine) {