Palav/dosh
1
0
Fork 0
Code Issues Pull Requests Actions 58 Packages Projects Releases 51 Wiki Activity

Retire terminal forwarding stream ids

Browse Source
This commit is contained in:
DuProcess
2026-07-10 22:23:34 -04:00
parent 5ca3e98a5c
commit 262dfaca04
2 changed files with 286 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/bin/dosh-server.rs
+136 -4
View File
@@ -47,6 +47,7 @@ use tokio::process::Command as TokioCommand;
use tokio::sync::mpsc;
const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024;
/// Persist a persistent session's screen to disk after this many bytes of output
/// have accumulated since the last mirror. Keeps the atomic write off the
@@ -399,6 +400,8 @@ struct ClientState {
stream_sent_data: HashMap<u64, BTreeMap<u64, PendingStreamChunk>>,
stream_next_recv_offset: HashMap<u64, u64>,
stream_recv_pending: HashMap<u64, BTreeMap<u64, Vec<u8>>>,
stream_retired: HashSet<u64>,
stream_retired_order: VecDeque<u64>,
/// Current transport key epoch (0 = original handshake key). Bumped on rekey.
epoch: u64,
/// When the current epoch began, for the wall-clock rekey trigger.
@@ -1201,6 +1204,8 @@ async fn handle_native_user_auth(
stream_sent_data: HashMap::new(),
stream_next_recv_offset: HashMap::new(),
stream_recv_pending: HashMap::new(),
stream_retired: HashSet::new(),
stream_retired_order: VecDeque::new(),
epoch: 0,
epoch_started: Instant::now(),
epoch_packets: 0,
@@ -1348,6 +1353,8 @@ async fn handle_bootstrap_attach(
stream_sent_data: HashMap::new(),
stream_next_recv_offset: HashMap::new(),
stream_recv_pending: HashMap::new(),
stream_retired: HashSet::new(),
stream_retired_order: VecDeque::new(),
epoch: 0,
epoch_started: Instant::now(),
epoch_packets: 0,
@@ -1482,6 +1489,8 @@ async fn handle_ticket_attach(
stream_sent_data: HashMap::new(),
stream_next_recv_offset: HashMap::new(),
stream_recv_pending: HashMap::new(),
stream_retired: HashSet::new(),
stream_retired_order: VecDeque::new(),
epoch: 0,
epoch_started: Instant::now(),
epoch_packets: 0,
@@ -1831,7 +1840,9 @@ async fn handle_stream_open(
}
client.endpoint = peer;
client.last_seen = Instant::now();
if client.opened_streams.contains(&open.stream_id) {
if client.stream_retired.contains(&open.stream_id) {
StreamOpenCheck::New(Err(anyhow!("stream {} is closed", open.stream_id)))
} else if client.opened_streams.contains(&open.stream_id) {
StreamOpenCheck::Duplicate
} else {
StreamOpenCheck::New(stream_open_allowed(&config, client, &open))
@@ -2138,7 +2149,7 @@ async fn handle_stream_open_reject(
}
client.endpoint = peer;
client.last_seen = Instant::now();
cleanup_client_stream(client, reject.stream_id);
retire_client_stream(client, reject.stream_id);
Ok(())
}
@@ -2196,7 +2207,7 @@ async fn handle_stream_close(
// Connection migration on any authenticated, fresh packet (spec §11).
client.endpoint = peer;
client.last_seen = Instant::now();
cleanup_client_stream(client, close.stream_id);
retire_client_stream(client, close.stream_id);
Ok(())
}
@@ -3288,6 +3299,34 @@ fn cleanup_client_stream(client: &mut ClientState, stream_id: u64) {
client.stream_recv_pending.remove(&stream_id);
}
fn client_has_stream_state(client: &ClientState, stream_id: u64) -> bool {
client.stream_writers.contains_key(&stream_id)
|| client.stream_pending_opens.contains_key(&stream_id)
|| client.opened_streams.contains(&stream_id)
|| client.stream_send_credit.contains_key(&stream_id)
|| client.stream_pending_data.contains_key(&stream_id)
|| client.stream_next_send_offset.contains_key(&stream_id)
|| client.stream_sent_data.contains_key(&stream_id)
|| client.stream_next_recv_offset.contains_key(&stream_id)
|| client.stream_recv_pending.contains_key(&stream_id)
}
fn retire_client_stream(client: &mut ClientState, stream_id: u64) {
let had_state = client_has_stream_state(client, stream_id);
cleanup_client_stream(client, stream_id);
if !had_state || STREAM_RETIRED_TOMBSTONES == 0 {
return;
}
if client.stream_retired.insert(stream_id) {
client.stream_retired_order.push_back(stream_id);
}
while client.stream_retired_order.len() > STREAM_RETIRED_TOMBSTONES {
if let Some(expired) = client.stream_retired_order.pop_front() {
client.stream_retired.remove(&expired);
}
}
}
fn ack_stream_data(client: &mut ClientState, stream_id: u64, received_offset: u64) {
let Some(sent) = client.stream_sent_data.get_mut(&stream_id) else {
return;
@@ -3348,7 +3387,7 @@ async fn send_stream_close_to_client(
{
let mut locked = state.lock().expect("server state poisoned");
if let Some(client) = locked.client_mut(&client_id) {
cleanup_client_stream(client, stream_id);
retire_client_stream(client, stream_id);
}
}
let body = protocol::to_body(&StreamClose { stream_id })?;
@@ -4047,6 +4086,8 @@ mod tests {
stream_sent_data: HashMap::new(),
stream_next_recv_offset: HashMap::new(),
stream_recv_pending: HashMap::new(),
stream_retired: HashSet::new(),
stream_retired_order: VecDeque::new(),
epoch: 0,
epoch_started: Instant::now(),
epoch_packets: 0,
@@ -4154,6 +4195,25 @@ mod tests {
assert!(client.stream_recv_pending.contains_key(&other_stream_id));
}
#[test]
fn retire_client_stream_tombstones_only_real_stream_state() {
let stream_id = 42;
let mut client = test_client_state([8u8; 32]);
client.opened_streams.insert(stream_id);
client
.stream_send_credit
.insert(stream_id, STREAM_INITIAL_WINDOW);
retire_client_stream(&mut client, stream_id);
retire_client_stream(&mut client, 99);
assert!(!client.opened_streams.contains(&stream_id));
assert!(!client.stream_send_credit.contains_key(&stream_id));
assert!(client.stream_retired.contains(&stream_id));
assert!(!client.stream_retired.contains(&99));
assert_eq!(client.stream_retired_order, VecDeque::from([stream_id]));
}
#[test]
fn client_index_stays_in_sync_with_session_clients() {
let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
@@ -4359,6 +4419,74 @@ mod tests {
assert_eq!(ok.stream_id, stream_id);
}
#[tokio::test]
async fn retired_stream_open_is_rejected_not_reopened() {
let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
let client_id = [15u8; 16];
let session_key = [16u8; 32];
let stream_id = 100;
let receiver = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let sender = Arc::new(UdpSocket::bind("127.0.0.1:0").await.unwrap());
let mut client = test_client_state(session_key);
client.endpoint = receiver.local_addr().unwrap();
client.stream_retired.insert(stream_id);
client.stream_retired_order.push_back(stream_id);
state.sessions.insert(
"test".to_string(),
Session {
pty: None,
parser: vt100::Parser::new(24, 80, 100),
clients: HashMap::from([(client_id, client)]),
output_seq: 0,
recent: VecDeque::new(),
empty_since: None,
holder_control: None,
persistent: false,
bytes_since_persist: 0,
last_persisted_seq: 0,
},
);
state.client_index.insert(client_id, "test".to_string());
let state = Arc::new(Mutex::new(state));
let raw = protocol::encode_encrypted(
PacketKind::StreamOpen,
client_id,
2,
0,
&session_key,
CLIENT_TO_SERVER,
&protocol::to_body(&StreamOpen {
stream_id,
target_host: "127.0.0.1".to_string(),
target_port: 9,
})
.unwrap(),
)
.unwrap();
let packet = protocol::decode(&raw).unwrap();
handle_stream_open(&state, &sender, receiver.local_addr().unwrap(), &packet)
.await
.unwrap();
let mut buf = [0u8; 2048];
let (n, _) = tokio::time::timeout(Duration::from_secs(1), receiver.recv_from(&mut buf))
.await
.unwrap()
.unwrap();
let packet = protocol::decode(&buf[..n]).unwrap();
assert_eq!(packet.header.kind, PacketKind::StreamOpenReject);
let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT).unwrap();
let reject: StreamOpenReject = protocol::from_body(&plain).unwrap();
assert_eq!(reject.stream_id, stream_id);
assert!(reject.reason.contains("closed"));
let locked = state.lock().unwrap();
let client = locked.sessions["test"].clients.get(&client_id).unwrap();
assert!(!client.opened_streams.contains(&stream_id));
}
#[tokio::test]
async fn pending_server_stream_open_is_retransmitted() {
let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
@@ -4546,6 +4674,8 @@ mod tests {
stream_sent_data: HashMap::new(),
stream_next_recv_offset: HashMap::new(),
stream_recv_pending: HashMap::new(),
stream_retired: HashSet::new(),
stream_retired_order: VecDeque::new(),
epoch: 0,
epoch_started: Instant::now(),
epoch_packets: 0,
@@ -4640,6 +4770,8 @@ mod tests {
stream_sent_data: HashMap::new(),
stream_next_recv_offset: HashMap::new(),
stream_recv_pending: HashMap::new(),
stream_retired: HashSet::new(),
stream_retired_order: VecDeque::new(),
epoch: 0,
epoch_started: Instant::now(),
epoch_packets: 0,