ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Add interactive client terminal transport

Browse source
This commit is contained in:
ms2k 2026-07-17 06:59:39 +09:00
commit 6816f0632e
4 changed files with 1454 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -20,8 +20,8 @@ use lazaret_storage_catalog::Catalog;
use crate::control::ControlSessionAdmission;
use crate::control::{
ControlError, NonInteractiveSessionExecutor, PublishedControlFace, SessionRegistry,
StagedControlFace, TrustAnchor,
ControlError, InteractiveSessionExecutor, NonInteractiveSessionExecutor, PublishedControlFace,
SessionRegistry, StagedControlFace, TrustAnchor,
};
use crate::domain_store::{DomainStoreMutation, HostCoordinatorDomainStore};
use crate::lifecycle::{
@ -73,6 +73,7 @@ pub struct ProductionBootstrapPlan<'a> {
pub bundle_digest: BundleDigest,
pub invoking_uid: u32,
pub non_interactive_executor: Arc<dyn NonInteractiveSessionExecutor>,
pub interactive_executor: Arc<dyn InteractiveSessionExecutor>,
pub(crate) runtime_staging: &'a crate::production::RuntimeStaging,
}
@ -361,7 +362,7 @@ impl<'a> ProductionLifecycleAdapter<'a> {
Participant::bare_intent(generation_id, ParticipantRole::Publication),
)])
.map_err(failure)?;
let staged = StagedControlFace::stage_with_executor(
let staged = StagedControlFace::stage_with_executors(
path_from_host(
self.plan
.configuration
@ -373,6 +374,7 @@ impl<'a> ProductionLifecycleAdapter<'a> {
anchor,
Arc::clone(&self.registry),
Arc::clone(&self.plan.non_interactive_executor),
Arc::clone(&self.plan.interactive_executor),
)
.map_err(failure)?;
let ownership = staged.recovery_ownership().map_err(failure)?;

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

@ -22,7 +22,8 @@ use lazaret_core::protocol::{
};
use lazaret_core::records::{
ClientEnrollmentRequest, ClientEnrollmentResponse, ControlRequest, ControlResponse,
DurableDecision, GenerationIdentity, NonInteractiveSessionMessage,
DurableDecision, GenerationIdentity, InteractiveSessionBatch, InteractiveSessionEvent,
InteractiveSessionMessage, InteractiveSessionRequest, NonInteractiveSessionMessage,
NonInteractiveSessionRequest, ProtocolCredential, ProtocolCredentialRole,
ProtocolPublicationIdentity, ProtocolSubjectId, ProtocolVersion, PublicationOwnership,
SandboxCredentialBundle, Session, SessionBinding, SessionExit, SessionPhase,
@ -834,6 +835,7 @@ pub struct PublishedControlFace {
server_config: Arc<ServerConfig>,
registry: Arc<SessionRegistry>,
executor: Arc<dyn NonInteractiveSessionExecutor>,
interactive_executor: Arc<dyn InteractiveSessionExecutor>,
}
pub trait NonInteractiveSessionExecutor: Send + Sync {
@ -858,6 +860,35 @@ impl NonInteractiveSessionExecutor for ExecutionUnavailable {
}
}
pub trait InteractiveControlSession: Send {
/// # Errors
/// Returns an exact-session terminal, prompt-routing, or transport failure.
fn exchange(
&mut self,
event: InteractiveSessionEvent,
) -> Result<InteractiveSessionBatch, ControlError>;
}
pub trait InteractiveSessionExecutor: Send + Sync {
/// # Errors
/// Returns an acquisition or exact-session identity failure.
fn start(
&self,
request: &InteractiveSessionRequest,
) -> Result<Box<dyn InteractiveControlSession>, ControlError>;
}
struct InteractiveExecutionUnavailable;
impl InteractiveSessionExecutor for InteractiveExecutionUnavailable {
fn start(
&self,
_request: &InteractiveSessionRequest,
) -> Result<Box<dyn InteractiveControlSession>, ControlError> {
Err(ControlError::InvalidRequest)
}
}
pub struct RunscNonInteractiveExecutor {
program: OsString,
global_arguments: Vec<OsString>,
@ -1009,6 +1040,7 @@ pub struct StagedControlFace {
server_config: Arc<ServerConfig>,
registry: Arc<SessionRegistry>,
executor: Arc<dyn NonInteractiveSessionExecutor>,
interactive_executor: Arc<dyn InteractiveSessionExecutor>,
}
impl StagedControlFace {
@ -1045,6 +1077,28 @@ impl StagedControlFace {
anchor: TrustAnchor,
registry: Arc<SessionRegistry>,
executor: Arc<dyn NonInteractiveSessionExecutor>,
) -> Result<Self, ControlError> {
Self::stage_with_executors(
runtime_domain_root,
uid,
anchor,
registry,
executor,
Arc::new(InteractiveExecutionUnavailable),
)
}
/// Stages a control face with generation-bound command and terminal execution.
///
/// # Errors
/// Refuses the same invalid staging conditions as [`Self::stage`].
pub fn stage_with_executors(
runtime_domain_root: &Path,
uid: u32,
anchor: TrustAnchor,
registry: Arc<SessionRegistry>,
executor: Arc<dyn NonInteractiveSessionExecutor>,
interactive_executor: Arc<dyn InteractiveSessionExecutor>,
) -> Result<Self, ControlError> {
if registry.generation_identity != anchor.generation_identity {
return Err(ControlError::StaleGeneration);
@ -1098,6 +1152,7 @@ impl StagedControlFace {
server_config,
registry,
executor,
interactive_executor,
})
}
@ -1156,6 +1211,7 @@ impl StagedControlFace {
server_config: Arc::clone(&self.server_config),
registry: Arc::clone(&self.registry),
executor: Arc::clone(&self.executor),
interactive_executor: Arc::clone(&self.interactive_executor),
})
}
}
@ -1379,6 +1435,33 @@ impl PublishedControlFace {
.expose()
}
/// Publishes one generation control face with command and terminal execution.
///
/// # Errors
/// Refuses the same invalid publication conditions as [`Self::publish`].
pub fn publish_with_executors(
runtime_domain_root: &Path,
uid: u32,
anchor: TrustAnchor,
store: Arc<HostCoordinatorDomainStore>,
executor: Arc<dyn NonInteractiveSessionExecutor>,
interactive_executor: Arc<dyn InteractiveSessionExecutor>,
) -> Result<Self, ControlError> {
let registry = Arc::new(SessionRegistry::new(
store,
anchor.generation_identity.clone(),
));
StagedControlFace::stage_with_executors(
runtime_domain_root,
uid,
anchor,
registry,
executor,
interactive_executor,
)?
.expose()
}
#[must_use]
pub fn publication_identity(&self) -> &ProtocolPublicationIdentity {
&self.publication_identity
@ -1534,32 +1617,50 @@ impl PublishedControlFace {
}
Err(error) => return Err(error),
};
let ControlRequest::Execute(request) = request else {
return Err(ControlError::InvalidRequest);
};
if request.session_id != session_id
|| request.argv.is_empty()
|| request.argv[0].is_empty()
|| request.argv.iter().any(|argument| argument.contains(&0))
{
return Err(ControlError::InvalidRequest);
}
let mut send_message = |message: NonInteractiveSessionMessage| {
if matches!(message, NonInteractiveSessionMessage::Exit(_)) {
return Err(ControlError::InvalidRequest);
match request {
ControlRequest::Execute(request) => {
validate_command(request.session_id, session_id, &request.argv)?;
let mut send_message = |message: NonInteractiveSessionMessage| {
if matches!(message, NonInteractiveSessionMessage::Exit(_)) {
return Err(ControlError::InvalidRequest);
}
require_non_interactive_session(&message, request.session_id)?;
write_frame(stream, &message)
};
let outcome = self.executor.execute(&request, &mut send_message)?;
require_session_id(outcome.session_id(), request.session_id)?;
write_frame(stream, &NonInteractiveSessionMessage::Exit(outcome))
}
ControlRequest::ExecuteInteractive(request) => {
validate_command(request.session_id, session_id, &request.argv)?;
if request.initial_size.session_id != session_id {
return Err(ControlError::InvalidRequest);
}
let mut terminal = self.interactive_executor.start(&request)?;
let mut event = InteractiveSessionEvent::Poll { session_id };
loop {
let batch = terminal.exchange(event)?;
require_interactive_batch(&batch, session_id)?;
let complete = batch.messages.iter().any(|message| {
matches!(
message,
InteractiveSessionMessage::Terminal(
lazaret_core::records::SessionTerminalMessage::Exit(_)
)
)
});
write_frame(stream, &batch)?;
if complete {
break Ok(());
}
event = read_frame(stream, MAX_CONTROL_BYTES)?;
require_interactive_event(&event, session_id)?;
}
}
ControlRequest::Enroll(_) | ControlRequest::Register(_) => {
Err(ControlError::InvalidRequest)
}
require_non_interactive_session(&message, request.session_id)?;
write_frame(stream, &message)
};
let outcome = self.executor.execute(&request, &mut send_message)?;
if outcome.session_id() != request.session_id {
return Err(ControlError::InvalidRequest);
}
require_non_interactive_session(
&NonInteractiveSessionMessage::Exit(outcome),
request.session_id,
)?;
write_frame(stream, &NonInteractiveSessionMessage::Exit(outcome))
})();
let disconnected = disconnect(session_id);
result.and(disconnected)
@ -1738,6 +1839,61 @@ fn require_non_interactive_session(
}
}
fn validate_command(
actual: SessionId,
expected: SessionId,
argv: &[Vec<u8>],
) -> Result<(), ControlError> {
require_session_id(actual, expected)?;
if argv.is_empty() || argv[0].is_empty() || argv.iter().any(|argument| argument.contains(&0)) {
return Err(ControlError::InvalidRequest);
}
Ok(())
}
fn require_session_id(actual: SessionId, expected: SessionId) -> Result<(), ControlError> {
if actual == expected {
Ok(())
} else {
Err(ControlError::InvalidRequest)
}
}
fn require_interactive_event(
event: &InteractiveSessionEvent,
expected: SessionId,
) -> Result<(), ControlError> {
let actual = match event {
InteractiveSessionEvent::Terminal(event) => match event {
lazaret_core::records::SessionTerminalEvent::Input(event) => event.session_id,
lazaret_core::records::SessionTerminalEvent::Resize(event) => event.session_id,
lazaret_core::records::SessionTerminalEvent::Signal(event) => event.session_id,
lazaret_core::records::SessionTerminalEvent::Close(event) => event.session_id,
},
InteractiveSessionEvent::PromptDecision { session_id, .. }
| InteractiveSessionEvent::Poll { session_id } => *session_id,
};
require_session_id(actual, expected)
}
fn require_interactive_batch(
batch: &InteractiveSessionBatch,
expected: SessionId,
) -> Result<(), ControlError> {
require_session_id(batch.session_id, expected)?;
for message in &batch.messages {
let actual = match message {
InteractiveSessionMessage::Terminal(message) => match message {
lazaret_core::records::SessionTerminalMessage::Output(output) => output.session_id,
lazaret_core::records::SessionTerminalMessage::Exit(exit) => exit.session_id(),
},
InteractiveSessionMessage::Prompt(prompt) => prompt.session_id,
};
require_session_id(actual, expected)?;
}
Ok(())
}
fn leaf_parameters(
common_name: &str,
usage: ExtendedKeyUsagePurpose,

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

@ -16,7 +16,7 @@ use lazaret_core::records::{
Generation, GenerationIdentity, HostProcessOwnershipSlot, IpAddress, PreparedArtifact, Session,
};
use lazaret_linux::filesystem::{mount_identity, namespace_identity};
use lazaret_linux::process::ProcessOwnershipContext;
use lazaret_linux::process::{ProcessOwnershipContext, current_process_identity};
use lazaret_runtime_supervisor::network::NetworkProbeRequest;
use lazaret_runtime_supervisor::sandbox::{
RootProcess, SandboxBuildRequest, SandboxBundle, build_sandbox_bundle,
@ -37,6 +37,7 @@ use crate::bootstrap::{
};
use crate::control::RunscNonInteractiveExecutor;
use crate::domain_store::HostCoordinatorDomainStore;
use crate::terminal::RunscInteractiveSessionExecutor;
use crate::waiter::{BootstrapWaitOutcome, BootstrapWaiterFace};
const NETWORK_NAMESPACE_PATH: &str = "/proc/self/ns/net";
@ -349,6 +350,17 @@ impl<'a> ProductionColdStart<'a> {
)
.map_err(failure)?,
),
interactive_executor: Arc::new(
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)?,
),
runtime_staging: &self.staging,
})
}

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

@ -14,11 +14,13 @@ use lazaret_core::identity::{
PidfsProcessIdentityV1, SessionId,
};
use lazaret_core::records::{
DomainStore, DurableDecision, GuestProcessIdentity, ParticipantOwnership, ParticipantRole,
ParticipantState, PersistedHostProcessOwnership, ProcessOwnershipCommitAck, PtyIdentity,
SessionExit, SessionPhase, SessionTerminal, SessionTerminalEvent, SessionTerminalFields,
SessionTerminalMessage, SessionTerminalState, StagedProcessOwnership, TerminalOutput,
TerminalResize, TerminalSignalKind,
DomainStore, DurableDecision, GuestProcessIdentity, InteractiveSessionBatch,
InteractiveSessionEvent, InteractiveSessionMessage, InteractiveSessionRequest,
ParticipantOwnership, ParticipantRole, ParticipantState, PersistedHostProcessOwnership,
ProcessOwnershipCommitAck, ProtocolVersion, PtyIdentity, SessionExit, SessionPhase,
SessionTerminal, SessionTerminalEvent, SessionTerminalFields, SessionTerminalMessage,
SessionTerminalState, StagedProcessOwnership, TerminalOutput, TerminalResize,
TerminalSignalKind,
};
use lazaret_linux::process::{
ChildStandardIo, ProcessExit, ProcessOwnershipContext, RetainedProcess, TargetCommand,
@ -28,6 +30,7 @@ use rustix::fs::{OFlags, fcntl_getfl, fcntl_setfl, fstat};
use rustix::pty::{OpenptFlags, grantpt, ioctl_tiocgptpeer, openpt, unlockpt};
use rustix::termios::{Winsize, tcsetwinsize};
use crate::control::{ControlError, InteractiveControlSession, InteractiveSessionExecutor};
use crate::domain_store::{DomainStoreError, DomainStoreMutation, HostCoordinatorDomainStore};
use crate::sandbox_init::{PRIVATE_SESSION_EXEC_ROLE, PRIVATE_SESSION_IDENTITY_ROLE};
@ -406,6 +409,175 @@ pub struct SessionTerminalEngine {
active: HashMap<SessionId, ActiveTerminal>,
}
pub struct RunscInteractiveSessionExecutor {
engine: Arc<Mutex<SessionTerminalEngine>>,
domain_id: DomainId,
generation_id: GenerationId,
coordinator_id: CoordinatorId,
container_id: ContainerId,
host_coordinator: PidfsProcessIdentityV1,
runsc_program: OsString,
runsc_global_arguments: Vec<OsString>,
runtime_root: PathBuf,
}
impl RunscInteractiveSessionExecutor {
/// Creates the generation-bound control adapter for the coordinator-owned terminal engine.
///
/// # Errors
/// Refuses a run command without one exact `run`/bundle/container suffix.
pub fn new(
store: Arc<HostCoordinatorDomainStore>,
generation: &lazaret_core::records::GenerationIdentity,
container_id: ContainerId,
host_coordinator: PidfsProcessIdentityV1,
command: &lazaret_runtime_supervisor::sandbox::RunscCommand,
runtime_root: PathBuf,
) -> Result<Self, TerminalError> {
let run = command
.arguments()
.iter()
.position(|argument| argument == "run")
.ok_or(TerminalError::AuthorityMismatch)?;
if run + 3 != command.arguments().len() {
return Err(TerminalError::AuthorityMismatch);
}
Ok(Self {
engine: Arc::new(Mutex::new(SessionTerminalEngine::new(
store,
Arc::new(RunscSessionTerminalFactory),
Arc::new(NoActivePrompt),
))),
domain_id: generation.domain_id,
generation_id: generation.generation_id,
coordinator_id: generation.coordinator_id,
container_id,
host_coordinator,
runsc_program: command.program().to_owned(),
runsc_global_arguments: command.arguments()[..run].to_vec(),
runtime_root,
})
}
}
impl InteractiveSessionExecutor for RunscInteractiveSessionExecutor {
fn start(
&self,
request: &InteractiveSessionRequest,
) -> Result<Box<dyn InteractiveControlSession>, ControlError> {
if request.protocol_version != ProtocolVersion::V1
|| request.initial_size.session_id != request.session_id
{
return Err(ControlError::InvalidRequest);
}
let acquisition = TerminalAcquisition {
domain_id: self.domain_id,
generation_id: self.generation_id,
coordinator_id: self.coordinator_id,
session_id: request.session_id,
container_id: self.container_id,
host_coordinator: self.host_coordinator,
runsc_program: self.runsc_program.clone(),
runsc_global_arguments: self.runsc_global_arguments.clone(),
internal_pid_file: self.runtime_root.join(format!(
"session-{}.pid",
hex_session(request.session_id.as_bytes())
)),
argv: request.argv.clone(),
};
let mut engine = self.engine.lock().map_err(|_| ControlError::LockPoisoned)?;
engine
.acquire(&acquisition)
.map_err(|error| terminal_control_error(&error))?;
if let Err(error) = engine.handle(
self.generation_id,
SessionTerminalEvent::Resize(request.initial_size),
) {
let _ = engine.lose(
request.session_id,
self.generation_id,
TerminalLoss::ClientDisconnected,
);
return Err(terminal_control_error(&error));
}
drop(engine);
Ok(Box::new(RunscInteractiveControlSession {
engine: Arc::clone(&self.engine),
generation_id: self.generation_id,
session_id: request.session_id,
}))
}
}
struct RunscInteractiveControlSession {
engine: Arc<Mutex<SessionTerminalEngine>>,
generation_id: GenerationId,
session_id: SessionId,
}
impl InteractiveControlSession for RunscInteractiveControlSession {
fn exchange(
&mut self,
event: InteractiveSessionEvent,
) -> Result<InteractiveSessionBatch, ControlError> {
let mut engine = self.engine.lock().map_err(|_| ControlError::LockPoisoned)?;
let mut messages = match event {
InteractiveSessionEvent::Terminal(event) => engine
.handle(self.generation_id, event)
.map_err(|error| terminal_control_error(&error))?,
InteractiveSessionEvent::Poll { session_id } if session_id == self.session_id => {
Vec::new()
}
InteractiveSessionEvent::PromptDecision { .. }
| InteractiveSessionEvent::Poll { .. } => return Err(ControlError::InvalidRequest),
};
if !messages
.iter()
.any(|message| matches!(message, SessionTerminalMessage::Exit(_)))
{
messages.extend(
engine
.poll(self.session_id, self.generation_id)
.map_err(|error| terminal_control_error(&error))?,
);
}
Ok(InteractiveSessionBatch {
session_id: self.session_id,
messages: messages
.into_iter()
.map(InteractiveSessionMessage::Terminal)
.collect(),
})
}
}
impl Drop for RunscInteractiveControlSession {
fn drop(&mut self) {
if let Ok(mut engine) = self.engine.lock()
&& engine.contains(self.session_id)
{
let _ = engine.lose(
self.session_id,
self.generation_id,
TerminalLoss::ClientDisconnected,
);
}
}
}
fn terminal_control_error(error: &TerminalError) -> ControlError {
ControlError::Lifecycle(error.to_string())
}
fn hex_session(bytes: &[u8]) -> String {
use std::fmt::Write as _;
bytes.iter().fold(String::new(), |mut output, byte| {
write!(output, "{byte:02x}").expect("writing to a String cannot fail");
output
})
}
impl SessionTerminalEngine {
#[must_use]
pub fn new(