use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64};
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use rmux_os::identity::UserIdentity;
use rmux_proto::{
CreateWebShareRequest, ErrorResponse, KillPaneRequest, KillSessionRequest, KillWindowRequest,
NewWindowRequest, PaneInputRequest, PaneResizeRequest, PaneSelectRequest, PaneTarget,
PaneTargetRef, RenameWindowRequest, ResizePaneAdjustment, Response, RmuxError,
SelectWindowRequest, SessionId, SessionName, SplitDirection, SplitWindowRequest,
SplitWindowTarget, WebShareRequest, WebShareScope, WindowTarget,
};
use tokio::sync::{mpsc, watch};
use super::attach_support::{attach_target_for_session, AttachRegistration, ClientFlags};
use super::pane_support::resolve_pane_target_ref;
use super::RequestHandler;
use crate::outer_terminal::OuterTerminalContext;
use crate::pane_io::{self, AttachControl, LiveAttachInputContext, PaneOutputReceiver};
use crate::pane_terminal_lookup::pane_id_for_target;
use crate::server_access::current_owner_uid;
use crate::web::{ResolvedCreateWebShareRequest, WebSessionTarget, WebShareAccess, WebShareTarget};
use rmux_core::{input::mode, PaneId};
const WEB_ATTACH_PID_BASE: u32 = 0x8000_0000;
#[path = "handler_web_snapshot.rs"]
mod snapshot;
#[path = "handler_web_stream.rs"]
mod stream;
use snapshot::{overlay_pane_lines, session_content_geometry, snapshot_ansi_lines, WebSessionView};
pub(crate) use snapshot::{WebPaneSnapshot, WebSessionSnapshot};
pub(crate) use stream::{
WebPaneStream, WebSessionAttachEvent, WebSessionAttachReader, WebSessionStream, WebShareStream,
};
impl RequestHandler {
#[cfg(test)]
pub(crate) async fn open_web_share(
&self,
token: &str,
pin: Option<&str>,
) -> Result<WebShareStream, RmuxError> {
let token_id = crate::web::SecretHashForCrypto::from_secret(token).token_id();
self.open_web_share_token_id(&token_id, pin).await
}
pub(crate) fn web_listener(&self) -> rmux_proto::WebShareListener {
self.web_shares.listener()
}
pub(crate) fn mark_web_listener_available(&self) {
self.web_shares.mark_listener_available();
}
pub(crate) fn mark_web_listener_unavailable(&self, reason: impl Into<String>) {
self.web_shares.mark_listener_unavailable(reason);
}
pub(crate) async fn ensure_web_share_listener_running(&self) -> Result<(), RmuxError> {
if self.web_shares.listener_available() {
return Ok(());
}
let _start_guard = self.web_listener_start.lock().await;
if self.web_shares.listener_available() {
return Ok(());
}
if let Err(error) = crate::web::spawn(Arc::new(self.clone())).await {
return Err(RmuxError::Server(format!(
"web-share listener unavailable: {error}"
)));
}
Ok(())
}
pub(in crate::handler) async fn handle_web_share(&self, request: WebShareRequest) -> Response {
let response = match request {
WebShareRequest::Create(request) => {
if let Err(error) = self.ensure_web_share_listener_running().await {
return Response::Error(ErrorResponse { error });
}
let request = match self.resolve_create_web_share(request).await {
Ok(request) => request,
Err(error) => return Response::Error(ErrorResponse { error }),
};
if let Err(error) = self.web_shares.validate_create_options(request.request()) {
return Response::Error(ErrorResponse { error });
}
let request = match self.start_web_share_tunnel(request).await {
Ok(request) => request,
Err(error) => return Response::Error(ErrorResponse { error }),
};
let expiry_kill_target = request.expiry_kill_target();
match self.web_shares.create(request) {
Ok(created) => {
self.spawn_web_share_expiry_task(
created.share_id.clone(),
created.expires_at_unix,
expiry_kill_target,
);
Ok(rmux_proto::WebShareResponse::Created(created))
}
Err(error) => Err(error),
}
}
WebShareRequest::Config(request) => {
if let Err(error) = self.ensure_web_share_listener_running().await {
return Response::Error(ErrorResponse { error });
}
self.web_shares.handle(WebShareRequest::Config(request))
}
other => self.web_shares.handle(other),
};
match response {
Ok(response) => Response::WebShare(response),
Err(error) => Response::Error(ErrorResponse { error }),
}
}
pub(crate) async fn open_web_share_token_id(
&self,
token_id: &str,
pin: Option<&str>,
) -> Result<WebShareStream, RmuxError> {
let access = self.web_shares.connect_token_id(token_id, pin).await?;
self.open_web_share_access(access).await
}
pub(crate) fn web_share_pre_auth_token(
&self,
token_id: &str,
origin: &str,
) -> Option<(crate::web::SecretHashForCrypto, bool)> {
self.web_shares.pre_auth_token(token_id, origin)
}
pub(in crate::handler) fn prune_web_session(&self, removed: Option<(SessionName, SessionId)>) {
if let Some((name, id)) = removed {
self.web_shares.remove_targets_for_sessions(&[(name, id)]);
}
}
async fn open_web_share_access(
&self,
access: WebShareAccess,
) -> Result<WebShareStream, RmuxError> {
match access.target().clone() {
WebShareTarget::Pane(target) => {
let target = self.stable_web_target(&target).await?;
let (snapshot, output) = self.web_resnapshot(&target).await?;
let revoke_rx = access.revoke_receiver();
Ok(WebShareStream::Pane(Box::new(WebPaneStream {
access,
output,
revoke_rx,
snapshot,
target,
})))
}
WebShareTarget::Session(session_target) => {
let stream = self.open_web_session_share(access, session_target).await?;
Ok(WebShareStream::Session(Box::new(stream)))
}
}
}
async fn open_web_session_share(
&self,
access: WebShareAccess,
session_target: WebSessionTarget,
) -> Result<WebSessionStream, RmuxError> {
let session_target = self.current_web_session_target(&session_target).await?;
let (server_transport, client_stream) = pane_io::in_process_attach_pair();
let attach_pid = self.allocate_web_attach_pid().await?;
let mut flags = ClientFlags::default();
let can_write = access.is_operator();
if !can_write {
flags = flags.with_read_only();
}
let terminal_context = OuterTerminalContext::default();
let (control_tx, control_rx) = mpsc::unbounded_channel::<AttachControl>();
let closing = Arc::new(AtomicBool::new(false));
let persistent_overlay_epoch = Arc::new(AtomicU64::new(0));
let attached_count = self
.active_attach
.lock()
.await
.attached_count(session_target.name());
let (session_target, target, snapshot) = {
let state = self.state.lock().await;
let session = state
.sessions
.session_by_id(session_target.id())
.ok_or_else(|| session_not_found_web(session_target.name()))?;
let current_target = WebSessionTarget::new(session.name().clone(), session.id());
let target = attach_target_for_session(
&state,
current_target.name(),
attached_count,
&terminal_context,
)?;
let snapshot = web_session_snapshot_from_state(
&state,
¤t_target,
target.render_frame.clone(),
&HashMap::new(),
)?;
(current_target.clone(), target, snapshot)
};
let attach_id = self
.register_attach_with_access(
attach_pid,
session_target.name().clone(),
AttachRegistration {
control_tx,
closing: closing.clone(),
persistent_overlay_epoch: persistent_overlay_epoch.clone(),
terminal_context,
flags,
uid: current_owner_uid(),
user: UserIdentity::Uid(current_owner_uid()),
can_write,
client_size: None,
},
)
.await;
let (_shutdown_tx, shutdown_rx) = watch::channel(());
let task_handler = self.clone();
tokio::spawn(async move {
let _keep_shutdown_open = _shutdown_tx;
let result = pane_io::forward_attach(
server_transport,
target,
Vec::new(),
shutdown_rx,
control_rx,
closing,
persistent_overlay_epoch,
LiveAttachInputContext {
handler: Arc::new(task_handler.clone()),
attach_pid,
},
)
.await;
task_handler.finish_attach(attach_pid, attach_id).await;
if let Err(error) = result {
tracing::debug!(attach_pid, "web session attach ended: {error}");
}
});
let revoke_rx = access.revoke_receiver();
let (reader, writer) = tokio::io::split(client_stream);
Ok(WebSessionStream {
access,
attach_pid,
revoke_rx,
target: session_target,
snapshot,
writer,
reader: Some(WebSessionAttachReader::new(reader)),
})
}
pub(crate) async fn web_session_snapshot_with_scrolls(
&self,
session_target: &WebSessionTarget,
scrolls: &HashMap<PaneId, usize>,
) -> Result<WebSessionSnapshot, RmuxError> {
let session_target = self.current_web_session_target(session_target).await?;
let attached_count = self
.active_attach
.lock()
.await
.attached_count(session_target.name());
let terminal_context = OuterTerminalContext::default();
let state = self.state.lock().await;
let session = state
.sessions
.session_by_id(session_target.id())
.ok_or_else(|| session_not_found_web(session_target.name()))?;
let target =
attach_target_for_session(&state, session.name(), attached_count, &terminal_context)?;
web_session_snapshot_from_state(&state, &session_target, target.render_frame, scrolls)
}
#[cfg(test)]
pub(crate) async fn web_session_snapshot(
&self,
session_target: &WebSessionTarget,
) -> Result<WebSessionSnapshot, RmuxError> {
self.web_session_snapshot_with_scrolls(session_target, &HashMap::new())
.await
}
pub(crate) async fn web_resnapshot(
&self,
target: &PaneTargetRef,
) -> Result<(WebPaneSnapshot, PaneOutputReceiver), RmuxError> {
let (pane_output, transcript) = {
let state = self.state.lock().await;
let target = resolve_pane_target_ref(&state, target)?;
let pane_output = state.pane_output_for_target(
target.session_name(),
target.window_index(),
target.pane_index(),
)?;
let transcript = state.transcript_handle(&target)?;
(pane_output, transcript)
};
let (output_sequence, snapshot) = pane_output.capture_with_next_sequence(|| {
let transcript = match transcript.lock() {
Ok(transcript) => transcript,
Err(poisoned) => poisoned.into_inner(),
};
let screen = transcript.clone_screen();
let size = screen.size();
let (cursor_col, cursor_row) = screen.cursor_position();
let (scroll_top, scroll_bottom) = screen.scroll_region();
WebPaneSnapshot {
cols: size.cols,
rows: size.rows,
output_sequence: 0,
ansi_lines: snapshot_ansi_lines(&screen),
cursor_row: cursor_row.min(u32::from(size.rows.saturating_sub(1))) as u16,
cursor_col: cursor_col.min(u32::from(size.cols.saturating_sub(1))) as u16,
cursor_visible: screen.mode() & mode::MODE_CURSOR != 0,
mode_bits: screen.mode(),
cursor_style: screen.cursor_style(),
alternate: screen.is_alternate(),
scroll_top,
scroll_bottom,
}
});
let snapshot = WebPaneSnapshot {
output_sequence,
..snapshot
};
let output = pane_output.subscribe_from_sequence(output_sequence);
Ok((snapshot, output))
}
pub(crate) async fn web_send_text(
&self,
target: &PaneTargetRef,
text: String,
) -> Result<(), RmuxError> {
let response = self
.handle_pane_input_ref(PaneInputRequest {
target: target.clone(),
keys: vec![text],
literal: true,
})
.await;
response_to_result(response)
}
pub(crate) async fn web_send_key(
&self,
target: &PaneTargetRef,
key: String,
) -> Result<(), RmuxError> {
let response = self
.handle_pane_input_ref(PaneInputRequest {
target: target.clone(),
keys: vec![key],
literal: false,
})
.await;
response_to_result(response)
}
pub(crate) async fn web_session_logout(
&self,
session_target: &WebSessionTarget,
) -> Result<(), RmuxError> {
let session_target = self.current_web_session_target(session_target).await?;
let response = self
.handle_kill_session(KillSessionRequest {
target: session_target.name().clone(),
kill_all_except_target: false,
clear_alerts: false,
})
.await;
response_to_result(response)
}
pub(crate) async fn web_session_select_pane(
&self,
session_target: &WebSessionTarget,
pane_id: PaneId,
) -> Result<(), RmuxError> {
let session_target = self.current_web_session_target(session_target).await?;
let response = self
.handle_pane_select_ref(PaneSelectRequest {
target: PaneTargetRef::by_id(session_target.name().clone(), pane_id),
title: None,
})
.await;
response_to_result(response)
}
pub(crate) async fn web_session_resize_pane(
&self,
session_target: &WebSessionTarget,
pane_id: PaneId,
adjustment: ResizePaneAdjustment,
) -> Result<(), RmuxError> {
let session_target = self.current_web_session_target(session_target).await?;
let response = self
.handle_pane_resize_ref(PaneResizeRequest {
target: PaneTargetRef::by_id(session_target.name().clone(), pane_id),
adjustment,
})
.await;
response_to_result(response)
}
pub(crate) async fn web_session_split_pane(
&self,
session_target: &WebSessionTarget,
requester_pid: u32,
direction: SplitDirection,
) -> Result<(), RmuxError> {
let session_target = self.current_web_session_target(session_target).await?;
let response = self
.handle_split_window(
requester_pid,
SplitWindowRequest {
target: SplitWindowTarget::Session(session_target.name().clone()),
direction,
before: false,
environment: None,
},
)
.await;
response_to_result(response)
}
pub(crate) async fn web_session_new_window(
&self,
session_target: &WebSessionTarget,
requester_pid: u32,
) -> Result<(), RmuxError> {
let session_target = self.current_web_session_target(session_target).await?;
let response = self
.handle_new_window(
requester_pid,
NewWindowRequest {
target: session_target.name().clone(),
name: None,
detached: false,
environment: None,
command: None,
process_command: None,
start_directory: None,
target_window_index: None,
insert_at_target: false,
},
)
.await;
response_to_result(response)
}
pub(crate) async fn web_session_kill_active_pane(
&self,
session_target: &WebSessionTarget,
) -> Result<(), RmuxError> {
let target = self.web_session_active_pane_target(session_target).await?;
let response = self
.handle_kill_pane(KillPaneRequest {
target,
kill_all_except: false,
})
.await;
response_to_result(response)
}
pub(crate) async fn web_session_select_window(
&self,
session_target: &WebSessionTarget,
window_index: u32,
) -> Result<(), RmuxError> {
let session_target = self.current_web_session_target(session_target).await?;
let response = self
.handle_select_window(SelectWindowRequest {
target: WindowTarget::with_window(session_target.name().clone(), window_index),
})
.await;
response_to_result(response)
}
pub(crate) async fn web_session_rename_window(
&self,
session_target: &WebSessionTarget,
window_index: u32,
name: String,
) -> Result<(), RmuxError> {
let session_target = self.current_web_session_target(session_target).await?;
let response = self
.handle_rename_window(RenameWindowRequest {
target: WindowTarget::with_window(session_target.name().clone(), window_index),
name,
})
.await;
response_to_result(response)
}
pub(crate) async fn web_session_kill_window(
&self,
session_target: &WebSessionTarget,
window_index: u32,
) -> Result<(), RmuxError> {
let session_target = self.current_web_session_target(session_target).await?;
let response = self
.handle_kill_window(KillWindowRequest {
target: WindowTarget::with_window(session_target.name().clone(), window_index),
kill_all_others: false,
})
.await;
response_to_result(response)
}
fn spawn_web_share_expiry_task(
&self,
share_id: String,
expires_at_unix: Option<u64>,
kill_target: Option<WebSessionTarget>,
) {
let Some(expires_at_unix) = expires_at_unix else {
return;
};
let handler = self.clone();
tokio::spawn(async move {
tokio::time::sleep(duration_until_unix(expires_at_unix)).await;
let Some(expired) = handler
.wait_for_web_share_expiry(&share_id, expires_at_unix)
.await
else {
return;
};
tracing::info!(share_id = %expired.share_id, "web_share_expired");
let target = kill_target.or(expired.kill_session);
if let Some(target) = target {
if let Err(error) = handler.web_session_logout(&target).await {
tracing::debug!(
share_id = %expired.share_id,
session = %target.name(),
"web-share expiry session kill skipped: {error}"
);
}
}
});
}
async fn wait_for_web_share_expiry(
&self,
share_id: &str,
expires_at_unix: u64,
) -> Option<crate::web::ExpiredWebShare> {
let retry_until = UNIX_EPOCH
.checked_add(Duration::from_secs(expires_at_unix))
.and_then(|deadline| deadline.checked_add(Duration::from_secs(1)))?;
loop {
if let Some(expired) = self.web_shares.expire_if_due(share_id) {
return Some(expired);
}
if SystemTime::now() >= retry_until {
return None;
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
}
async fn resolve_create_web_share(
&self,
request: CreateWebShareRequest,
) -> Result<ResolvedCreateWebShareRequest, rmux_proto::RmuxError> {
let state = self.state.lock().await;
let target = match &request.scope {
WebShareScope::Pane(raw_target) => {
let target = resolve_pane_target_ref(&state, raw_target)?;
let pane_id = pane_id_for_target(
&state.sessions,
target.session_name(),
target.window_index(),
target.pane_index(),
)?;
WebShareTarget::pane(PaneTargetRef::by_id(target.session_name().clone(), pane_id))
}
WebShareScope::Session(session_name) => {
let session = state
.sessions
.session(session_name)
.ok_or_else(|| session_not_found_web(session_name))?;
WebShareTarget::session(session.name().clone(), session.id())
}
};
Ok(ResolvedCreateWebShareRequest::new(request, target))
}
async fn start_web_share_tunnel(
&self,
request: ResolvedCreateWebShareRequest,
) -> Result<ResolvedCreateWebShareRequest, rmux_proto::RmuxError> {
let provider = match request.tunnel_provider() {
Some(provider) => provider.to_owned(),
None => return Ok(request),
};
if request.public_base_url().is_some() {
return Err(RmuxError::Server(
"web-share --tunnel-url and --tunnel-provider are mutually exclusive".to_owned(),
));
}
let settings = self.web_shares.settings();
let tunnel = crate::web::start_tunnel_provider(&provider, &settings).await?;
Ok(request.with_tunnel(tunnel))
}
async fn stable_web_target(&self, target: &PaneTargetRef) -> Result<PaneTargetRef, RmuxError> {
let state = self.state.lock().await;
let target = resolve_pane_target_ref(&state, target)?;
let pane_id = pane_id_for_target(
&state.sessions,
target.session_name(),
target.window_index(),
target.pane_index(),
)?;
Ok(PaneTargetRef::by_id(target.session_name().clone(), pane_id))
}
pub(crate) async fn web_target_alive(&self, target: &PaneTargetRef) -> bool {
let state = self.state.lock().await;
resolve_pane_target_ref(&state, target).is_ok()
}
pub(crate) async fn web_session_alive(&self, session_target: &WebSessionTarget) -> bool {
self.current_web_session_target(session_target)
.await
.is_ok()
}
async fn current_web_session_target(
&self,
session_target: &WebSessionTarget,
) -> Result<WebSessionTarget, RmuxError> {
let state = self.state.lock().await;
state
.sessions
.session_by_id(session_target.id())
.map(|session| WebSessionTarget::new(session.name().clone(), session.id()))
.ok_or_else(|| session_not_found_web(session_target.name()))
}
async fn web_session_active_pane_target(
&self,
session_target: &WebSessionTarget,
) -> Result<PaneTarget, RmuxError> {
let session_target = self.current_web_session_target(session_target).await?;
let state = self.state.lock().await;
let session = state
.sessions
.session_by_id(session_target.id())
.ok_or_else(|| session_not_found_web(session_target.name()))?;
let window_index = session.active_window_index();
let pane_index = session.active_pane_index();
Ok(PaneTarget::with_window(
session_target.name().clone(),
window_index,
pane_index,
))
}
async fn allocate_web_attach_pid(&self) -> Result<u32, RmuxError> {
for _ in 0..1024 {
let id = self.allocate_connection_id();
let candidate = WEB_ATTACH_PID_BASE | (id as u32 & !WEB_ATTACH_PID_BASE);
if !self
.active_attach
.lock()
.await
.by_pid
.contains_key(&candidate)
{
return Ok(candidate);
}
}
Err(RmuxError::Server(
"failed to allocate web attach client id".to_owned(),
))
}
}
fn duration_until_unix(expires_at_unix: u64) -> Duration {
let Some(deadline) = UNIX_EPOCH.checked_add(Duration::from_secs(expires_at_unix)) else {
return Duration::ZERO;
};
deadline
.duration_since(SystemTime::now())
.unwrap_or(Duration::ZERO)
}
fn session_not_found_web(session_name: &SessionName) -> RmuxError {
RmuxError::Server(format!("can't find session: {session_name}"))
}
fn response_to_result(response: Response) -> Result<(), RmuxError> {
match response {
Response::Error(error) => Err(error.error),
_ => Ok(()),
}
}
fn web_session_snapshot_from_state(
state: &crate::pane_terminals::HandlerState,
session_target: &WebSessionTarget,
mut frame: Vec<u8>,
scrolls: &HashMap<PaneId, usize>,
) -> Result<WebSessionSnapshot, RmuxError> {
let session = state
.sessions
.session_by_id(session_target.id())
.ok_or_else(|| session_not_found_web(session_target.name()))?;
let window = session.window();
let mut view = WebSessionView::new(window.size());
let active_window = session.active_window_index();
let active_pane = window.active_pane_index();
for (index, window) in session.windows() {
view.add_window(*index, window.name(), *index == active_window);
}
let panes = if window.is_zoomed() {
window.active_pane().into_iter().collect::<Vec<_>>()
} else {
window.panes().iter().collect::<Vec<_>>()
};
let mut active_mode_bits = mode::MODE_CURSOR | mode::MODE_WRAP;
let mut active_cursor_style = 0u32;
for pane in panes {
if pane.index() == active_pane {
if let Some(screen) = state.pane_render_screen(session.name(), pane.id()) {
active_mode_bits = screen.mode();
active_cursor_style = screen.cursor_style();
}
}
let Some(geometry) = session_content_geometry(pane.geometry(), window.size()) else {
continue;
};
let scrollback = state
.pane_scrollback_view(
session.name(),
pane.id(),
scrolls.get(&pane.id()).copied().unwrap_or_default(),
)
.ok_or_else(|| RmuxError::Server(format!("missing pane transcript: {}", pane.id())))?;
if scrollback.scroll_offset > 0 {
overlay_pane_lines(&mut frame, geometry, &scrollback.ansi_lines);
}
view.add_pane(
pane.id(),
geometry,
pane.index() == active_pane,
scrollback.history_size,
scrollback.scroll_offset,
scrollback.alternate_on,
);
}
Ok(WebSessionSnapshot::new(
window.size(),
frame,
view,
active_mode_bits,
active_cursor_style,
))
}
#[cfg(test)]
#[path = "handler_web_tests.rs"]
mod tests;