naps62/maestro
Archived
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(maestro): reject structured PTY access

Browse Source
This commit is contained in:
Miguel Palhas
2026-09-05 10:20:41 +01:00
parent d3e3010245
commit f765d415d6
5 changed files with 213 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/maestro-daemon/src/session.rs
+90 -23
View File
@@ -169,6 +169,8 @@ pub enum SessionError {
/// a stray keystroke on an attach socket, which is dropped.
#[error("session is no longer running: {0}")]
NotRunning(Uuid),
#[error("session is not a terminal session")]
NotTerminal(Uuid),
#[error("daemon is shutting down")]
ShuttingDown,
/// The kill was refused (#79): the session's worktree holds unsaved work.
@@ -205,6 +207,7 @@ impl SessionError {
SessionError::NotFound(_) => "not_found",
SessionError::InvalidEnv(_) => "invalid_env",
SessionError::NotRunning(_) => "not_running",
SessionError::NotTerminal(_) => "not_terminal",
SessionError::ShuttingDown => "shutting_down",
SessionError::KillGuarded(_) => "kill_guarded",
};
@@ -879,11 +882,17 @@ impl SessionManager {
}
}
/// Subscribe to a session's PTY broadcast. `None` if the session is
/// unknown to this daemon.
pub async fn subscribe_pty(&self, id: Uuid) -> Option<broadcast::Receiver<Vec<u8>>> {
/// Subscribe to a terminal session's PTY broadcast.
pub async fn subscribe_pty(
&self,
id: Uuid,
) -> Result<broadcast::Receiver<Vec<u8>>, SessionError> {
let sessions = self.inner.sessions.lock().await;
sessions.get(&id).map(|r| r.bridge.subscribe())
let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
if record.session.view != SessionView::Terminal {
return Err(SessionError::NotTerminal(id));
}
Ok(record.bridge.subscribe())
}
/// Subscribe to a session's PTY broadcast and return the replay buffer
@@ -897,11 +906,14 @@ impl SessionManager {
pub async fn subscribe_pty_with_replay(
&self,
id: Uuid,
) -> Option<(Vec<u8>, broadcast::Receiver<Vec<u8>>)> {
) -> Result<(Vec<u8>, broadcast::Receiver<Vec<u8>>), SessionError> {
let tmux = self.inner.tmux.clone();
let (result, updated) = {
let mut sessions = self.inner.sessions.lock().await;
let rec = sessions.get_mut(&id)?;
let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
if rec.session.view != SessionView::Terminal {
return Err(SessionError::NotTerminal(id));
}
rec.session.last_focused_at = Some(Utc::now());
let updated = rec.session.clone();
// For a live session, tmux's own screen beats the byte tail: the
@@ -932,7 +944,7 @@ impl SessionManager {
.inner
.events
.send(CatalogEvent::SessionUpdated { session: updated });
Some(result)
Ok(result)
}
pub async fn has(&self, id: Uuid) -> bool {
@@ -945,6 +957,15 @@ impl SessionManager {
sessions.get(&id).map(|r| r.session.status)
}
async fn terminal_status(&self, id: Uuid) -> Result<SessionStatus, SessionError> {
let sessions = self.inner.sessions.lock().await;
let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
if record.session.view != SessionView::Terminal {
return Err(SessionError::NotTerminal(id));
}
Ok(record.session.status)
}
/// Whether the natural-exit settle step (#82) has finished for this
/// session: `Some(true)` once applied or refused, `Some(false)` while
/// still pending, `None` if the session is unknown. `Dead` is observable
@@ -1626,10 +1647,9 @@ impl SessionManager {
/// its pane is gone, so this is a no-op rather than a tmux failure the
/// client would have to interpret.
pub async fn send_keys(&self, id: Uuid, bytes: &[u8]) -> Result<(), SessionError> {
match self.status(id).await {
None => return Err(SessionError::NotFound(id)),
Some(SessionStatus::Dead) => return Ok(()),
Some(SessionStatus::Running) => {}
match self.terminal_status(id).await? {
SessionStatus::Dead => return Ok(()),
SessionStatus::Running => {}
}
let tmux = self.inner.tmux.clone();
let bytes = bytes.to_vec();
@@ -1645,10 +1665,9 @@ impl SessionManager {
/// asked for text to land somewhere, and a silent success would let a
/// script believe an agent had been told something it never heard.
pub async fn send_input(&self, id: Uuid, text: &str, submit: bool) -> Result<(), SessionError> {
match self.status(id).await {
None => return Err(SessionError::NotFound(id)),
Some(SessionStatus::Dead) => return Err(SessionError::NotRunning(id)),
Some(SessionStatus::Running) => {}
match self.terminal_status(id).await? {
SessionStatus::Dead => return Err(SessionError::NotRunning(id)),
SessionStatus::Running => {}
}
let tmux = self.inner.tmux.clone();
let bytes = text.as_bytes().to_vec();
@@ -1668,10 +1687,9 @@ impl SessionManager {
/// Resize a session's window. No-op for a dead session, as with
/// [`Self::send_keys`].
pub async fn resize(&self, id: Uuid, cols: u16, rows: u16) -> Result<(), SessionError> {
match self.status(id).await {
None => return Err(SessionError::NotFound(id)),
Some(SessionStatus::Dead) => return Ok(()),
Some(SessionStatus::Running) => {}
match self.terminal_status(id).await? {
SessionStatus::Dead => return Ok(()),
SessionStatus::Running => {}
}
let tmux = self.inner.tmux.clone();
tmux_blocking(move || tmux.resize_window(id, cols, rows))
@@ -1703,6 +1721,7 @@ impl SessionManager {
cols: u16,
rows: u16,
) -> Result<(), SessionError> {
self.terminal_status(id).await?;
let target = self
.inner
.viewports
@@ -2807,13 +2826,15 @@ mod tests {
async fn manager_with_structured_stream() -> (
tempfile::TempDir,
SessionManager,
Arc<FakePanes>,
uuid::Uuid,
Arc<tokio::sync::Mutex<AcpEventStream>>,
uuid::Uuid,
) {
let state = tempfile::tempdir().unwrap();
let panes = Arc::new(FakePanes::new());
let manager = SessionManager::new(
Arc::new(FakePanes::new()),
panes.clone(),
SidecarStore::new(state.path().join("sessions")),
WorktreeManager::new(None),
state.path().join("fifos"),
@@ -2848,12 +2869,12 @@ mod tests {
record.focused_since = Some(std::time::Instant::now() - FOCUS_CLEAR_HOLD);
record.acp_events = Some(events.clone());
}
(state, manager, structured.id, events, terminal.id)
(state, manager, panes, structured.id, events, terminal.id)
}
#[tokio::test]
async fn blocked_structured_activity_read_does_not_block_the_catalog() {
let (state, manager, structured_id, events, terminal_id) =
let (state, manager, _panes, structured_id, events, terminal_id) =
manager_with_structured_stream().await;
let held_stream = events.lock().await;
let activity_pass = run_structured_activity_pass(&manager.inner);
@@ -2898,9 +2919,55 @@ mod tests {
assert_eq!(structured.activity, ActivityState::Idle);
}
#[tokio::test]
async fn structured_sessions_reject_every_pty_operation_before_panes() {
let (_state, manager, panes, structured_id, _events, terminal_id) =
manager_with_structured_stream().await;
let before = panes.calls();
assert!(matches!(
manager.subscribe_pty(structured_id).await,
Err(super::SessionError::NotTerminal(id)) if id == structured_id
));
assert!(matches!(
manager.subscribe_pty_with_replay(structured_id).await,
Err(super::SessionError::NotTerminal(id)) if id == structured_id
));
assert!(matches!(
manager.send_keys(structured_id, b"ignored").await,
Err(super::SessionError::NotTerminal(id)) if id == structured_id
));
assert!(matches!(
manager.send_input(structured_id, "ignored", false).await,
Err(super::SessionError::NotTerminal(id)) if id == structured_id
));
assert!(matches!(
manager.resize(structured_id, 80, 24).await,
Err(super::SessionError::NotTerminal(id)) if id == structured_id
));
assert!(matches!(
manager
.set_viewport(structured_id, manager.attach_viewport(), 80, 24)
.await,
Err(super::SessionError::NotTerminal(id)) if id == structured_id
));
let after = panes.calls();
assert_eq!(after, before, "structured PTY operations touched panes");
manager
.send_input(terminal_id, "works", false)
.await
.expect("terminal input remains available");
manager
.resize(terminal_id, 100, 30)
.await
.expect("terminal resize remains available");
}
#[tokio::test]
async fn delayed_activity_from_a_replaced_stream_is_discarded() {
let (_state, manager, structured_id, old_events, _terminal_id) =
let (_state, manager, _panes, structured_id, old_events, _terminal_id) =
manager_with_structured_stream().await;
let replacement = Arc::new(tokio::sync::Mutex::new(stream(
manager