Expand native auth and benchmark gates

src/bin/dosh-server.rs

@@ -878,9 +878,11 @@ async fn handle_native_client_hello(
             .hello
             .supported_user_key_algorithms
             .iter()
-            .any(|algorithm| algorithm == "ssh-ed25519")
+            .any(|algorithm| dosh::native::is_supported_user_signature_algorithm(algorithm))
         {
-            Err(anyhow!("native auth requires ssh-ed25519"))
+            Err(anyhow!(
+                "native auth requires a supported user key algorithm"
+            ))
         } else {
             let current_user = std::env::var("USER").unwrap_or_else(|_| "unknown".to_string());
             if req.hello.requested_user != current_user {
@@ -3266,6 +3268,96 @@ mod tests {
         assert_eq!(data.bytes, b"hello");
     }
 
+    #[tokio::test]
+    async fn blocked_stream_data_does_not_block_terminal_frames() {
+        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
+        let client_id = [8u8; 16];
+        let session_key = [10u8; 32];
+        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 endpoint = receiver.local_addr().unwrap();
+        state.sessions.insert(
+            "test".to_string(),
+            Session {
+                pty: None,
+                parser: vt100::Parser::new(24, 80, 100),
+                clients: HashMap::from([(
+                    client_id,
+                    ClientState {
+                        endpoint,
+                        mode: "read-write".to_string(),
+                        session_key,
+                        last_acked: 0,
+                        replay: ReplayWindow::default(),
+                        send_seq: 1,
+                        cols: 80,
+                        rows: 24,
+                        last_seen: Instant::now(),
+                        pending: VecDeque::new(),
+                        allowed_forwardings: Vec::new(),
+                        stream_writers: HashMap::new(),
+                        opened_streams: HashSet::from([42]),
+                        stream_send_credit: HashMap::from([(42, 0)]),
+                        stream_pending_data: HashMap::new(),
+                        epoch: 0,
+                        epoch_started: Instant::now(),
+                        epoch_packets: 0,
+                        previous_session_key: None,
+                    },
+                )]),
+                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));
+
+        queue_or_send_stream_data_to_client(
+            &state,
+            &sender,
+            client_id,
+            42,
+            b"blocked-forward-bytes".to_vec(),
+        )
+        .await
+        .unwrap();
+        {
+            let locked = state.lock().unwrap();
+            let client = locked.sessions["test"].clients.get(&client_id).unwrap();
+            assert_eq!(client.stream_pending_data[&42].len(), 1);
+            assert_eq!(client.stream_send_credit[&42], 0);
+        }
+
+        broadcast_output(
+            &state,
+            &sender,
+            PtyOutput {
+                session: "test".to_string(),
+                bytes: b"terminal-priority".to_vec(),
+                exited: false,
+            },
+        )
+        .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::Frame);
+        let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT).unwrap();
+        let frame: Frame = protocol::from_body(&plain).unwrap();
+        assert_eq!(frame.bytes, b"terminal-priority");
+    }
+
     #[test]
     fn remote_non_loopback_bind_requires_explicit_config() {
         let config = ServerConfig::default();