263 lines
10 KiB
Rust
263 lines
10 KiB
Rust
use super::{
|
|
Arc, BTreeMap, ContainerId, ControlError, CoordinatorId, DomainId, EnvironmentValue,
|
|
GenerationId, HashMap, HostCoordinatorDomainStore, HostCoordinatorTerminalRecorder,
|
|
InteractiveControlSession, InteractiveSessionBatch, InteractiveSessionEvent,
|
|
InteractiveSessionExecutor, InteractiveSessionMessage, InteractiveSessionRequest, Mutex,
|
|
OsString, PathBuf, PidfsProcessIdentityV1, PromptRouteCancellation, ProtocolVersion,
|
|
RunscSessionTerminalFactory, SessionId, SessionPromptTransport, SessionTerminal,
|
|
SessionTerminalEvent, SessionTerminalFactory, SessionTerminalMessage, SessionTerminalProcess,
|
|
TerminalAcquisition, TerminalError, TerminalLoss, TerminalRoute, VecDeque,
|
|
};
|
|
|
|
pub(super) struct ActiveTerminal {
|
|
pub(super) generation_id: GenerationId,
|
|
pub(super) request: TerminalAcquisition,
|
|
pub(super) route: TerminalRoute,
|
|
pub(super) record: SessionTerminal,
|
|
pub(super) process: Box<dyn SessionTerminalProcess>,
|
|
pub(super) releasing: bool,
|
|
pub(super) pending_input: VecDeque<u8>,
|
|
}
|
|
|
|
pub struct SessionTerminalEngine {
|
|
pub(super) recorder: HostCoordinatorTerminalRecorder,
|
|
pub(super) factory: Arc<dyn SessionTerminalFactory>,
|
|
pub(super) cancellation: Arc<dyn PromptRouteCancellation>,
|
|
pub(super) active: HashMap<SessionId, ActiveTerminal>,
|
|
}
|
|
|
|
pub struct RunscInteractiveSessionExecutor {
|
|
engine: Arc<Mutex<SessionTerminalEngine>>,
|
|
prompt_transport: Arc<SessionPromptTransport>,
|
|
domain_id: DomainId,
|
|
generation_id: GenerationId,
|
|
coordinator_id: CoordinatorId,
|
|
container_id: ContainerId,
|
|
host_coordinator: PidfsProcessIdentityV1,
|
|
runsc_program: OsString,
|
|
runsc_global_arguments: Vec<OsString>,
|
|
environment: BTreeMap<String, EnvironmentValue>,
|
|
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,
|
|
environment: BTreeMap<String, EnvironmentValue>,
|
|
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);
|
|
}
|
|
let prompt_transport = Arc::new(SessionPromptTransport::new(Arc::clone(&store)));
|
|
Ok(Self {
|
|
engine: Arc::new(Mutex::new(SessionTerminalEngine::new(
|
|
store,
|
|
Arc::new(RunscSessionTerminalFactory),
|
|
prompt_transport.clone(),
|
|
))),
|
|
prompt_transport,
|
|
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(),
|
|
environment,
|
|
runtime_root,
|
|
})
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn prompt_transport(&self) -> Arc<SessionPromptTransport> {
|
|
Arc::clone(&self.prompt_transport)
|
|
}
|
|
}
|
|
|
|
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(),
|
|
environment: self.environment.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),
|
|
prompt_transport: Arc::clone(&self.prompt_transport),
|
|
generation_id: self.generation_id,
|
|
session_id: request.session_id,
|
|
}))
|
|
}
|
|
}
|
|
|
|
pub(super) struct RunscInteractiveControlSession {
|
|
pub(super) engine: Arc<Mutex<SessionTerminalEngine>>,
|
|
pub(super) prompt_transport: Arc<SessionPromptTransport>,
|
|
pub(super) generation_id: GenerationId,
|
|
pub(super) 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(SessionTerminalEvent::Resize(resize)) => {
|
|
if self
|
|
.prompt_transport
|
|
.retain_prompt_resize(resize)
|
|
.map_err(|error| terminal_control_error(&error))?
|
|
{
|
|
Vec::new()
|
|
} else {
|
|
engine
|
|
.handle(self.generation_id, SessionTerminalEvent::Resize(resize))
|
|
.map_err(|error| terminal_control_error(&error))?
|
|
}
|
|
}
|
|
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 {
|
|
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.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))?;
|
|
if let Some(resize) = decision.resumed_size {
|
|
engine
|
|
.handle(self.generation_id, SessionTerminalEvent::Resize(resize))
|
|
.map_err(|error| terminal_control_error(&error))?;
|
|
}
|
|
}
|
|
self.prompt_transport
|
|
.publish_decision(session_id, decision)
|
|
.map_err(|error| terminal_control_error(&error))?;
|
|
Vec::new()
|
|
}
|
|
InteractiveSessionEvent::PromptDecision { .. }
|
|
| 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
|
|
.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
|
|
})
|
|
}
|