#![allow(clippy::collapsible_if)] use std::fs; use std::io::{Read, Write}; use std::net::{TcpListener, TcpStream, UdpSocket}; use std::os::unix::net::{UnixListener, UnixStream}; use std::path::{Path, PathBuf}; use std::process::{Child, Command, Stdio}; use std::sync::mpsc; use std::thread; use std::time::Duration; use base64::Engine; use base64::engine::general_purpose::STANDARD; use dosh::auth::{build_bootstrap, load_or_create_server_secret}; use dosh::config::load_server_config; use dosh::crypto; use dosh::native::{host_public_key, load_or_create_host_key, ssh_ed25519_public_blob, trust_host}; use dosh::protocol::{ self, AttachOk, AttachReject, BootstrapAttachRequest, CLIENT_TO_SERVER, Frame, Input, PacketKind, Resize, ResumeRequest, SERVER_TO_CLIENT, TicketAttachBody, TicketAttachEnvelope, TicketAttachOkEnvelope, }; use ed25519_dalek::{Signer, SigningKey, VerifyingKey}; fn free_udp_port() -> u16 { let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); socket.local_addr().unwrap().port() } fn free_tcp_port() -> u16 { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); listener.local_addr().unwrap().port() } fn write_server_config(dir: &tempfile::TempDir, port: u16) -> std::path::PathBuf { write_server_config_with_timeout(dir, port, 30) } fn write_server_config_with_remote_forwarding( dir: &tempfile::TempDir, port: u16, ) -> std::path::PathBuf { let config = write_server_config(dir, port); let mut raw = fs::read_to_string(&config).unwrap(); raw.push_str("allow_remote_forwarding = true\n"); fs::write(&config, raw).unwrap(); config } fn write_server_config_with_agent_forwarding( dir: &tempfile::TempDir, port: u16, ) -> std::path::PathBuf { let config = write_server_config(dir, port); let mut raw = fs::read_to_string(&config).unwrap(); raw.push_str("allow_agent_forwarding = true\n"); fs::write(&config, raw).unwrap(); config } fn write_server_config_with_timeout( dir: &tempfile::TempDir, port: u16, client_timeout_secs: u64, ) -> std::path::PathBuf { let config_dir = dir.path().join(".config/dosh"); fs::create_dir_all(&config_dir).unwrap(); let config = config_dir.join("server.toml"); fs::write( &config, format!( r#" port = {port} bind = "127.0.0.1" scrollback = 5000 auth_ttl_secs = 30 attach_ticket_ttl_secs = 3600 allow_attach_tickets = true client_timeout_secs = {client_timeout_secs} retransmit_window = 256 default_input_mode = "read-write" prewarm_sessions = ["default"] create_on_attach = true shell = "/bin/sh" sessions_dir = "{sessions}" secret_path = "{secret}" host_key = "{host_key}" authorized_keys = ["{authorized_keys}"] persist_sessions = false "#, sessions = dir.path().join("sessions").display(), secret = dir.path().join("secret").display(), host_key = dir.path().join("host_key").display(), authorized_keys = dir.path().join("authorized_keys").display(), ), ) .unwrap(); config } fn write_server_config_with_output_interval( dir: &tempfile::TempDir, port: u16, output_frame_interval_ms: u64, ) -> std::path::PathBuf { let config = write_server_config(dir, port); let mut raw = fs::read_to_string(&config).unwrap(); raw.push_str(&format!( "output_frame_interval_ms = {output_frame_interval_ms}\n" )); fs::write(&config, raw).unwrap(); config } fn write_server_config_with_retransmit_window( dir: &tempfile::TempDir, port: u16, retransmit_window: u64, ) -> std::path::PathBuf { let config = write_server_config(dir, port); let mut raw = fs::read_to_string(&config).unwrap(); raw = raw.replace( "retransmit_window = 256\n", &format!("retransmit_window = {retransmit_window}\n"), ); fs::write(&config, raw).unwrap(); config } fn start_server(dir: &tempfile::TempDir, config: &std::path::Path) -> Child { let server = env!("CARGO_BIN_EXE_dosh-server"); let child = Command::new(server) .arg("serve") .arg("--config") .arg(config) .env("HOME", dir.path()) .stdout(Stdio::null()) .stderr(Stdio::null()) .spawn() .unwrap(); thread::sleep(Duration::from_millis(500)); child } fn attach_once(dir: &tempfile::TempDir, port: u16, cache: bool) -> std::process::Output { let client = env!("CARGO_BIN_EXE_dosh-client"); let mut cmd = Command::new(client); cmd.arg("--local-auth") .arg("--attach-only") .arg("-v") .arg("--session") .arg("default") .arg("--dosh-port") .arg(port.to_string()) .arg("local") .env("HOME", dir.path()); if !cache { cmd.arg("--no-cache"); } cmd.output().unwrap() } fn native_attach_once(dir: &tempfile::TempDir, port: u16) -> std::process::Output { let client = env!("CARGO_BIN_EXE_dosh-client"); Command::new(client) .arg("--auth") .arg("native") .arg("--attach-only") .arg("-v") .arg("--session") .arg("default") .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("local") .env("HOME", dir.path()) .output() .unwrap() } fn native_attach_once_with_agent( dir: &tempfile::TempDir, port: u16, agent_sock: &Path, ) -> std::process::Output { let client = env!("CARGO_BIN_EXE_dosh-client"); Command::new(client) .arg("--auth") .arg("native") .arg("--attach-only") .arg("-v") .arg("--session") .arg("default") .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("local") .env("HOME", dir.path()) .env("SSH_AUTH_SOCK", agent_sock) .output() .unwrap() } fn connect_with_retry(port: u16, timeout: Duration, child: &mut Child) -> TcpStream { let deadline = std::time::Instant::now() + timeout; loop { if let Some(status) = child.try_wait().unwrap() { let mut stderr = String::new(); if let Some(mut pipe) = child.stderr.take() { let _ = pipe.read_to_string(&mut stderr); } panic!( "client exited before local forward was ready: status={status}; stderr={stderr}" ); } match TcpStream::connect(("127.0.0.1", port)) { Ok(stream) => return stream, Err(err) if std::time::Instant::now() < deadline => { let _ = err; thread::sleep(Duration::from_millis(50)); } Err(err) => panic!("connect to local forward 127.0.0.1:{port}: {err}"), } } } fn socks5_connect(port: u16, target_port: u16, child: &mut Child) -> TcpStream { let mut stream = connect_with_retry(port, Duration::from_secs(5), child); stream .set_read_timeout(Some(Duration::from_secs(3))) .unwrap(); stream.write_all(&[5, 1, 0]).unwrap(); let mut response = [0u8; 2]; stream.read_exact(&mut response).unwrap(); assert_eq!(response, [5, 0]); let mut request = vec![5, 1, 0, 1, 127, 0, 0, 1]; request.extend_from_slice(&target_port.to_be_bytes()); stream.write_all(&request).unwrap(); let mut reply = [0u8; 10]; stream.read_exact(&mut reply).unwrap(); assert_eq!(reply[0], 5); assert_eq!(reply[1], 0); stream } fn start_echo_server() -> u16 { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let port = listener.local_addr().unwrap().port(); thread::spawn(move || { if let Ok((mut stream, _)) = listener.accept() { let mut buf = [0u8; 1024]; loop { match stream.read(&mut buf) { Ok(0) | Err(_) => break, Ok(n) => { if stream.write_all(&buf[..n]).is_err() { break; } } } } } }); port } fn start_slow_sink_server() -> u16 { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let port = listener.local_addr().unwrap().port(); thread::spawn(move || { if let Ok((mut stream, _)) = listener.accept() { let mut buf = [0u8; 8192]; loop { match stream.read(&mut buf) { Ok(0) | Err(_) => break, Ok(_) => thread::sleep(Duration::from_millis(20)), } } } }); port } fn connect_with_retry_no_child(port: u16, timeout: Duration) -> TcpStream { let deadline = std::time::Instant::now() + timeout; loop { match TcpStream::connect(("127.0.0.1", port)) { Ok(stream) => return stream, Err(err) if std::time::Instant::now() < deadline => { let _ = err; thread::sleep(Duration::from_millis(50)); } Err(err) => panic!("connect to local forward 127.0.0.1:{port}: {err}"), } } } fn write_native_client_auth(dir: &tempfile::TempDir, config_path: &std::path::Path) { write_native_client_auth_with_options(dir, config_path, ""); } fn write_native_client_auth_with_options( dir: &tempfile::TempDir, config_path: &std::path::Path, options: &str, ) { let ssh_dir = dir.path().join(".ssh"); fs::create_dir_all(&ssh_dir).unwrap(); let signing = ed25519_dalek::SigningKey::from_bytes(&[42u8; 32]); let keypair = ssh_key::private::Ed25519Keypair::from(&signing); let private = ssh_key::PrivateKey::new(ssh_key::private::KeypairData::from(keypair), "test").unwrap(); private .write_openssh_file(&ssh_dir.join("id_ed25519"), ssh_key::LineEnding::LF) .unwrap(); let public = private.public_key().to_openssh().unwrap(); let line = if options.is_empty() { public } else { format!("{options} {public}") }; fs::write(dir.path().join("authorized_keys"), format!("{line}\n")).unwrap(); let config = load_server_config(Some(config_path.to_path_buf())).unwrap(); let host_signing = load_or_create_host_key(&config).unwrap(); let known_hosts = dir.path().join(".config/dosh/known_hosts"); trust_host( &known_hosts, "local", &host_public_key(&host_signing), "test", false, ) .unwrap(); } fn write_native_agent_auth( dir: &tempfile::TempDir, config_path: &std::path::Path, signing: &SigningKey, ) { let public = VerifyingKey::from(signing).to_bytes(); fs::write( dir.path().join("authorized_keys"), format!( "ssh-ed25519 {} agent-key\n", STANDARD.encode(ssh_ed25519_public_blob(&public)) ), ) .unwrap(); let config = load_server_config(Some(config_path.to_path_buf())).unwrap(); let host_signing = load_or_create_host_key(&config).unwrap(); let known_hosts = dir.path().join(".config/dosh/known_hosts"); trust_host( &known_hosts, "local", &host_public_key(&host_signing), "test", false, ) .unwrap(); } fn start_fake_ssh_agent(dir: &tempfile::TempDir, signing: SigningKey) -> PathBuf { let socket_path = dir.path().join("agent.sock"); let public = VerifyingKey::from(&signing).to_bytes(); let key_blob = ssh_ed25519_public_blob(&public); let listener = UnixListener::bind(&socket_path).unwrap(); thread::spawn(move || fake_ssh_agent(listener, key_blob, signing)); socket_path } fn fake_ssh_agent(listener: UnixListener, key_blob: Vec, signing: SigningKey) { let (mut stream, _) = listener.accept().unwrap(); let request = read_agent_packet(&mut stream).unwrap(); assert_eq!(request, vec![11]); let mut response = Vec::new(); response.push(12); response.extend_from_slice(&1u32.to_be_bytes()); write_ssh_string(&mut response, &key_blob); write_ssh_string(&mut response, b"integration-agent-key"); write_agent_packet(&mut stream, &response).unwrap(); let request = read_agent_packet(&mut stream).unwrap(); let mut cursor = request.as_slice(); assert_eq!(read_u8(&mut cursor), 13); assert_eq!(read_ssh_string(&mut cursor), key_blob.as_slice()); let transcript = read_ssh_string(&mut cursor); assert_eq!(read_u32(&mut cursor), 0); assert!(cursor.is_empty()); let signature = signing.sign(transcript).to_bytes(); let mut signature_blob = Vec::new(); write_ssh_string(&mut signature_blob, b"ssh-ed25519"); write_ssh_string(&mut signature_blob, &signature); let mut response = Vec::new(); response.push(14); write_ssh_string(&mut response, &signature_blob); write_agent_packet(&mut stream, &response).unwrap(); } /// A fake ssh-agent that serves repeated connections and only answers /// REQUEST_IDENTITIES (enough to prove the forwarded socket reaches the real /// local agent). Returns the comment string it advertises so the test can match. fn start_fake_ssh_agent_looping(socket_path: &Path, comment: &'static str) { let signing = SigningKey::from_bytes(&[77u8; 32]); let public = VerifyingKey::from(&signing).to_bytes(); let key_blob = ssh_ed25519_public_blob(&public); let listener = UnixListener::bind(socket_path).unwrap(); thread::spawn(move || { for stream in listener.incoming() { let Ok(mut stream) = stream else { break }; let key_blob = key_blob.clone(); thread::spawn(move || { while let Ok(request) = read_agent_packet(&mut stream) { // Only the identities request (11) is exercised. if request.first() == Some(&11) { let mut response = Vec::new(); response.push(12); // IDENTITIES_ANSWER response.extend_from_slice(&1u32.to_be_bytes()); write_ssh_string(&mut response, &key_blob); write_ssh_string(&mut response, comment.as_bytes()); if write_agent_packet(&mut stream, &response).is_err() { break; } } else { // Unknown request: reply AGENT_FAILURE (5). let _ = write_agent_packet(&mut stream, &[5]); } } }); } }); } fn read_agent_packet(stream: &mut UnixStream) -> std::io::Result> { let mut len = [0u8; 4]; stream.read_exact(&mut len)?; let len = u32::from_be_bytes(len) as usize; let mut payload = vec![0u8; len]; stream.read_exact(&mut payload)?; Ok(payload) } fn write_agent_packet(stream: &mut UnixStream, payload: &[u8]) -> std::io::Result<()> { stream.write_all(&(payload.len() as u32).to_be_bytes())?; stream.write_all(payload) } fn read_u8(cursor: &mut &[u8]) -> u8 { let value = cursor[0]; *cursor = &cursor[1..]; value } fn read_u32(cursor: &mut &[u8]) -> u32 { let value = u32::from_be_bytes(cursor[..4].try_into().unwrap()); *cursor = &cursor[4..]; value } fn read_ssh_string<'a>(cursor: &mut &'a [u8]) -> &'a [u8] { let len = read_u32(cursor) as usize; let (value, rest) = cursor.split_at(len); *cursor = rest; value } fn write_ssh_string(out: &mut Vec, value: &[u8]) { out.extend_from_slice(&(value.len() as u32).to_be_bytes()); out.extend_from_slice(value); } fn direct_attach( config: &std::path::Path, port: u16, mode: &str, ) -> (std::net::UdpSocket, dosh::auth::BootstrapResponse, AttachOk) { direct_attach_session(config, port, mode, "default") } fn direct_attach_session( config: &std::path::Path, port: u16, mode: &str, session: &str, ) -> (std::net::UdpSocket, dosh::auth::BootstrapResponse, AttachOk) { let config = load_server_config(Some(config.to_path_buf())).unwrap(); let secret = load_or_create_server_secret(&config).unwrap(); let bootstrap = build_bootstrap( &config, &secret, "tester".to_string(), session.to_string(), mode.to_string(), (80, 24), crypto::random_12(), "127.0.0.1".to_string(), ) .unwrap(); let socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); socket .set_read_timeout(Some(Duration::from_secs(2))) .unwrap(); let req = BootstrapAttachRequest { bootstrap: bootstrap.clone(), cols: 80, rows: 24, requested_env: Vec::new(), }; let packet = protocol::encode_plain( PacketKind::BootstrapAttachRequest, [0u8; 16], 1, 0, &protocol::to_body(&req).unwrap(), ) .unwrap(); socket .send_to(&packet, format!("127.0.0.1:{port}")) .unwrap(); let mut buf = [0u8; 65535]; let (n, _) = socket.recv_from(&mut buf).unwrap(); let packet = protocol::decode(&buf[..n]).unwrap(); assert_eq!(packet.header.kind, PacketKind::AttachOk); let plain = protocol::decrypt_body(&packet, &bootstrap.session_key, SERVER_TO_CLIENT).unwrap(); let ok: AttachOk = protocol::from_body(&plain).unwrap(); (socket, bootstrap, ok) } fn ticket_attach_session( socket: &UdpSocket, port: u16, bootstrap: &dosh::auth::BootstrapResponse, session: &str, mode: &str, ) -> AttachOk { let client_nonce = crypto::random_12(); let request_key = crypto::hkdf32( &bootstrap.attach_ticket_psk, &client_nonce, b"dosh/ticket-attach-request/v1", ) .unwrap(); let body = protocol::to_body(&TicketAttachBody { session: session.to_string(), mode: mode.to_string(), cols: 80, rows: 24, requested_env: Vec::new(), }) .unwrap(); let ciphertext = crypto::seal( &request_key, &client_nonce, b"dosh-ticket-attach-request-v1", &body, ) .unwrap(); let envelope = TicketAttachEnvelope { ticket: bootstrap.attach_ticket.clone(), client_nonce, ciphertext, }; let packet = protocol::encode_plain( PacketKind::TicketAttachRequest, [0u8; 16], 1, 0, &protocol::to_body(&envelope).unwrap(), ) .unwrap(); socket .send_to(&packet, format!("127.0.0.1:{port}")) .unwrap(); let mut buf = [0u8; 65535]; let (n, _) = socket.recv_from(&mut buf).unwrap(); let packet = protocol::decode(&buf[..n]).unwrap(); assert_eq!(packet.header.kind, PacketKind::AttachOk); let envelope: TicketAttachOkEnvelope = protocol::from_body(&packet.body).unwrap(); let mut salt = Vec::with_capacity(24); salt.extend_from_slice(&client_nonce); salt.extend_from_slice(&envelope.server_nonce); let response_key = crypto::hkdf32( &bootstrap.attach_ticket_psk, &salt, b"dosh/ticket-attach-ok/v1", ) .unwrap(); let plain = crypto::open( &response_key, &envelope.server_nonce, b"dosh-ticket-attach-ok-v1", &envelope.ciphertext, ) .unwrap(); protocol::from_body(&plain).unwrap() } #[allow(clippy::too_many_arguments)] fn send_encrypted( socket: &UdpSocket, port: u16, kind: PacketKind, client_id: [u8; 16], seq: u64, ack: u64, key: &[u8; 32], body: &[u8], ) { let packet = protocol::encode_encrypted(kind, client_id, seq, ack, key, CLIENT_TO_SERVER, body).unwrap(); socket .send_to(&packet, format!("127.0.0.1:{port}")) .unwrap(); } fn recv_frame(socket: &UdpSocket, key: &[u8; 32]) -> Option<(protocol::Header, Frame)> { let mut buf = [0u8; 65535]; let (n, _) = socket.recv_from(&mut buf).ok()?; let packet = protocol::decode(&buf[..n]).ok()?; match packet.header.kind { PacketKind::Frame | PacketKind::ResumeOk => { let plain = protocol::decrypt_body(&packet, key, SERVER_TO_CLIENT).ok()?; let frame: Frame = protocol::from_body(&plain).ok()?; Some((packet.header, frame)) } _ => None, } } fn collect_frame_text(socket: &UdpSocket, key: &[u8; 32], millis: u64) -> String { socket .set_read_timeout(Some(Duration::from_millis(100))) .unwrap(); let deadline = std::time::Instant::now() + Duration::from_millis(millis); let mut text = String::new(); while std::time::Instant::now() < deadline { if let Some((_header, frame)) = recv_frame(socket, key) { text.push_str(&String::from_utf8_lossy(&frame.bytes)); } } socket .set_read_timeout(Some(Duration::from_secs(2))) .unwrap(); text } #[test] fn unknown_live_client_gets_reject() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); socket .set_read_timeout(Some(Duration::from_secs(2))) .unwrap(); let client_id = crypto::random_16(); let key = crypto::random_32(); let packet = protocol::encode_encrypted( PacketKind::Ping, client_id, 1, 0, &key, CLIENT_TO_SERVER, b"", ) .unwrap(); socket .send_to(&packet, format!("127.0.0.1:{port}")) .unwrap(); let mut buf = [0u8; 65535]; let (n, _) = socket.recv_from(&mut buf).unwrap(); let packet = protocol::decode(&buf[..n]).unwrap(); let reject: AttachReject = protocol::from_body(&packet.body).unwrap(); let _ = server.kill(); let _ = server.wait(); assert_eq!(packet.header.kind, PacketKind::AttachReject); assert_eq!(packet.header.conn_id, client_id); assert_eq!(reject.reason, "unknown client"); } #[test] fn disconnected_client_times_out_and_gets_reject() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config_with_timeout(&dir, port, 1); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); thread::sleep(Duration::from_secs(6)); send_encrypted( &socket, port, PacketKind::Ping, ok.client_id, 2, 0, &bootstrap.session_key, b"", ); let mut buf = [0u8; 65535]; let (n, _) = socket.recv_from(&mut buf).unwrap(); let packet = protocol::decode(&buf[..n]).unwrap(); let reject: AttachReject = protocol::from_body(&packet.body).unwrap(); let _ = server.kill(); let _ = server.wait(); assert_eq!(packet.header.kind, PacketKind::AttachReject); assert_eq!(packet.header.conn_id, ok.client_id); assert_eq!(reject.reason, "unknown client"); } #[test] fn known_client_with_stale_key_id_gets_reject_not_timeout() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (_bootstrap, ok) = { let (_socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); (bootstrap, ok) }; let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); socket .set_read_timeout(Some(Duration::from_secs(2))) .unwrap(); let stale_key = crypto::random_32(); let packet = protocol::encode_encrypted( PacketKind::Ping, ok.client_id, 99, 0, &stale_key, CLIENT_TO_SERVER, b"", ) .unwrap(); socket .send_to(&packet, format!("127.0.0.1:{port}")) .unwrap(); let mut buf = [0u8; 65535]; let (n, _) = socket.recv_from(&mut buf).unwrap(); let packet = protocol::decode(&buf[..n]).unwrap(); let reject: AttachReject = protocol::from_body(&packet.body).unwrap(); let _ = server.kill(); let _ = server.wait(); assert_eq!(packet.header.kind, PacketKind::AttachReject); assert_eq!(packet.header.conn_id, ok.client_id); assert_eq!(reject.reason, "unknown client"); } #[test] fn local_attach_only_smoke() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let output = attach_once(&dir, port, false); let _ = server.kill(); let _ = server.wait(); assert!( output.status.success(), "stderr={}", String::from_utf8_lossy(&output.stderr) ); assert!( String::from_utf8_lossy(&output.stderr).contains("terminal_ready"), "stderr={}", String::from_utf8_lossy(&output.stderr) ); } #[test] fn native_attach_only_smoke() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); write_native_client_auth(&dir, &config); let mut server = start_server(&dir, &config); let output = native_attach_once(&dir, port); let _ = server.kill(); let _ = server.wait(); let stderr = String::from_utf8_lossy(&output.stderr); assert!(output.status.success(), "stderr={stderr}"); assert!(stderr.contains("native_auth"), "stderr={stderr}"); assert!(stderr.contains("terminal_ready"), "stderr={stderr}"); } #[test] fn native_attach_enforces_authorized_keys_from_option() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); write_native_client_auth_with_options(&dir, &config, "from=\"10.99.0.0/16\""); let mut server = start_server(&dir, &config); let output = native_attach_once(&dir, port); let _ = server.kill(); let _ = server.wait(); let stderr = String::from_utf8_lossy(&output.stderr); assert!(!output.status.success(), "stderr={stderr}"); assert!(stderr.contains("from="), "stderr={stderr}"); } #[test] fn native_attach_only_smoke_uses_ssh_agent() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let signing = SigningKey::from_bytes(&[53u8; 32]); write_native_agent_auth(&dir, &config, &signing); let agent_sock = start_fake_ssh_agent(&dir, signing); let mut server = start_server(&dir, &config); let output = native_attach_once_with_agent(&dir, port, &agent_sock); let _ = server.kill(); let _ = server.wait(); let stderr = String::from_utf8_lossy(&output.stderr); assert!(output.status.success(), "stderr={stderr}"); assert!(stderr.contains("native_auth"), "stderr={stderr}"); assert!(stderr.contains("terminal_ready"), "stderr={stderr}"); } #[test] fn native_doctor_checks_auth_without_opening_terminal() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); write_native_client_auth(&dir, &config); let mut server = start_server(&dir, &config); let client = env!("CARGO_BIN_EXE_dosh-client"); let output = Command::new(client) .arg("doctor") .arg("--auth") .arg("native") .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("local") .env("HOME", dir.path()) .output() .unwrap(); let _ = server.kill(); let _ = server.wait(); let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); assert!(output.status.success(), "stdout={stdout}\nstderr={stderr}"); assert!(stdout.contains("[ok] native udp"), "stdout={stdout}"); assert!(stdout.contains("[ok] native auth"), "stdout={stdout}"); assert!(stdout.contains("[ok] server policy"), "stdout={stdout}"); assert!(stdout.contains("file_transfer=true"), "stdout={stdout}"); assert!(stdout.contains("exec_command=true"), "stdout={stdout}"); } #[test] fn native_local_forward_echo_smoke() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); write_native_client_auth(&dir, &config); let echo_port = start_echo_server(); let local_port = free_tcp_port(); let mut server = start_server(&dir, &config); let client_bin = env!("CARGO_BIN_EXE_dosh-client"); let mut client = Command::new(client_bin) .arg("--auth") .arg("native") .arg("--no-cache") .arg("-N") .arg("-L") .arg(format!("{local_port}:127.0.0.1:{echo_port}")) .arg("--session") .arg("default") .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("local") .env("HOME", dir.path()) .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::piped()) .spawn() .unwrap(); let mut stream = connect_with_retry(local_port, Duration::from_secs(5), &mut client); stream .set_read_timeout(Some(Duration::from_secs(3))) .unwrap(); stream.write_all(b"dosh-forward-ping").unwrap(); let mut buf = [0u8; 64]; let n = stream.read(&mut buf).unwrap(); let _ = client.kill(); let client_output = client.wait_with_output().unwrap(); let _ = server.kill(); let _ = server.wait(); assert_eq!(&buf[..n], b"dosh-forward-ping"); assert!( client_output.status.success() || client_output.status.code().is_none(), "stderr={}", String::from_utf8_lossy(&client_output.stderr) ); } #[test] fn native_local_forward_background_smoke() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); write_native_client_auth(&dir, &config); let echo_port = start_echo_server(); let local_port = free_tcp_port(); let pid_file = dir.path().join("background.pid"); let mut server = start_server(&dir, &config); let client_bin = env!("CARGO_BIN_EXE_dosh-client"); let output = Command::new(client_bin) .arg("--auth") .arg("native") .arg("--no-cache") .arg("-f") .arg("-N") .arg("-L") .arg(format!("{local_port}:127.0.0.1:{echo_port}")) .arg("--session") .arg("default") .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("local") .env("HOME", dir.path()) .env("DOSH_BACKGROUND_PID_FILE", &pid_file) .output() .unwrap(); assert!( output.status.success(), "stderr={}", String::from_utf8_lossy(&output.stderr) ); let mut stream = TcpStream::connect(("127.0.0.1", local_port)).unwrap(); stream .set_read_timeout(Some(Duration::from_secs(3))) .unwrap(); stream.write_all(b"dosh-background-forward-ping").unwrap(); let mut buf = [0u8; 64]; let n = stream.read(&mut buf).unwrap(); if let Ok(pid) = fs::read_to_string(&pid_file) { let _ = Command::new("kill").arg(pid.trim()).status(); } let _ = server.kill(); let _ = server.wait(); assert_eq!(&buf[..n], b"dosh-background-forward-ping"); } #[test] fn native_file_copy_recursive_round_trip() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); write_native_client_auth(&dir, &config); let mut server = start_server(&dir, &config); let src = dir.path().join("srcdir"); fs::create_dir_all(src.join("nested")).unwrap(); fs::write(src.join("root.txt"), b"root file\n").unwrap(); fs::write(src.join("nested/child.txt"), b"child file\n").unwrap(); let client_bin = env!("CARGO_BIN_EXE_dosh-client"); let upload = Command::new(client_bin) .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("cp") .arg("-r") .arg(&src) .arg("local:remote-copy") .env("HOME", dir.path()) .output() .unwrap(); assert!( upload.status.success(), "upload failed: stdout={} stderr={}", String::from_utf8_lossy(&upload.stdout), String::from_utf8_lossy(&upload.stderr) ); let remote_copy = Command::new(client_bin) .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("cp") .arg("-r") .arg("local:remote-copy") .arg("local:remote-copy-2") .env("HOME", dir.path()) .output() .unwrap(); assert!( remote_copy.status.success(), "remote copy failed: stdout={} stderr={}", String::from_utf8_lossy(&remote_copy.stdout), String::from_utf8_lossy(&remote_copy.stderr) ); let list = Command::new(client_bin) .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("ls") .arg("local:remote-copy-2") .env("HOME", dir.path()) .output() .unwrap(); assert!( list.status.success(), "ls failed: stdout={} stderr={}", String::from_utf8_lossy(&list.stdout), String::from_utf8_lossy(&list.stderr) ); let list_stdout = String::from_utf8_lossy(&list.stdout); assert!(list_stdout.contains("root.txt"), "stdout={list_stdout}"); assert!(list_stdout.contains("nested"), "stdout={list_stdout}"); let cat = Command::new(client_bin) .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("cat") .arg("local:remote-copy-2/root.txt") .env("HOME", dir.path()) .output() .unwrap(); assert!( cat.status.success(), "cat failed: stdout={} stderr={}", String::from_utf8_lossy(&cat.stdout), String::from_utf8_lossy(&cat.stderr) ); assert_eq!(String::from_utf8_lossy(&cat.stdout), "root file\n"); let downloaded = dir.path().join("downloaded"); let download = Command::new(client_bin) .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("cp") .arg("-r") .arg("local:remote-copy-2") .arg(&downloaded) .env("HOME", dir.path()) .output() .unwrap(); assert!( download.status.success(), "download failed: stdout={} stderr={}", String::from_utf8_lossy(&download.stdout), String::from_utf8_lossy(&download.stderr) ); assert_eq!( fs::read_to_string(downloaded.join("root.txt")).unwrap(), "root file\n" ); assert_eq!( fs::read_to_string(downloaded.join("nested/child.txt")).unwrap(), "child file\n" ); let remove = Command::new(client_bin) .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("rm") .arg("-r") .arg("local:remote-copy") .env("HOME", dir.path()) .output() .unwrap(); assert!( remove.status.success(), "rm failed: stdout={} stderr={}", String::from_utf8_lossy(&remove.stdout), String::from_utf8_lossy(&remove.stderr) ); let _ = server.kill(); let _ = server.wait(); } #[test] fn native_exec_command_smoke() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); write_native_client_auth(&dir, &config); let mut server = start_server(&dir, &config); let client_bin = env!("CARGO_BIN_EXE_dosh-client"); let output = Command::new(client_bin) .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("exec") .arg("local") .arg("printf out; printf err >&2") .env("HOME", dir.path()) .output() .unwrap(); let _ = server.kill(); let _ = server.wait(); assert!( output.status.success(), "exec failed: stdout={} stderr={}", String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr) ); assert_eq!(String::from_utf8_lossy(&output.stdout), "out"); assert_eq!(String::from_utf8_lossy(&output.stderr), "err"); } #[test] fn native_local_forward_bulk_load_does_not_delay_interactive_terminal() { use portable_pty::{CommandBuilder, NativePtySystem, PtySize, PtySystem}; let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); write_native_client_auth(&dir, &config); let sink_port = start_slow_sink_server(); let local_port = free_tcp_port(); let mut server = start_server(&dir, &config); let pty = NativePtySystem::default(); let pair = pty .openpty(PtySize { rows: 24, cols: 80, pixel_width: 0, pixel_height: 0, }) .unwrap(); let mut reader = pair.master.try_clone_reader().unwrap(); let mut writer = pair.master.take_writer().unwrap(); let client_bin = env!("CARGO_BIN_EXE_dosh-client"); let mut cmd = CommandBuilder::new(client_bin); let forward_arg = format!("{local_port}:127.0.0.1:{sink_port}"); let dosh_port = port.to_string(); cmd.args([ "--auth", "native", "--no-cache", "-L", &forward_arg, "--session", "load-session", "--dosh-host", "127.0.0.1", "--dosh-port", &dosh_port, "local", ]); cmd.env("HOME", dir.path().to_string_lossy().to_string()); let mut child = pair.slave.spawn_command(cmd).unwrap(); drop(pair.slave); let mut forward_stream = connect_with_retry_no_child(local_port, Duration::from_secs(5)); let bulk_writer = thread::spawn(move || { let chunk = vec![b'x'; 8192]; for _ in 0..128 { if forward_stream.write_all(&chunk).is_err() { break; } } }); let (seen_tx, seen_rx) = mpsc::channel(); let reader_handle = thread::spawn(move || { let mut buf = [0u8; 4096]; let mut output = Vec::new(); loop { match reader.read(&mut buf) { Ok(0) | Err(_) => break, Ok(n) => { output.extend_from_slice(&buf[..n]); if output .windows(b"DOSH_LOAD_TERMINAL_OK".len()) .any(|w| w == b"DOSH_LOAD_TERMINAL_OK") { let _ = seen_tx.send(String::from_utf8_lossy(&output).to_string()); break; } } } } }); thread::sleep(Duration::from_millis(250)); writer .write_all(b"printf DOSH_LOAD_TERMINAL_OK\\n\r") .unwrap(); writer.flush().unwrap(); let seen = seen_rx.recv_timeout(Duration::from_secs(3)); let _ = child.kill(); let _ = child.wait(); drop(writer); drop(pair.master); let _ = reader_handle.join(); let _ = bulk_writer.join(); let _ = server.kill(); let _ = server.wait(); assert!( seen.is_ok(), "interactive terminal output was delayed behind local-forward bulk load" ); } #[test] fn native_remote_forward_echo_smoke() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config_with_remote_forwarding(&dir, port); write_native_client_auth(&dir, &config); let echo_port = start_echo_server(); let remote_port = free_tcp_port(); let mut server = start_server(&dir, &config); let client_bin = env!("CARGO_BIN_EXE_dosh-client"); let mut client = Command::new(client_bin) .arg("--auth") .arg("native") .arg("--no-cache") .arg("-N") .arg("-R") .arg(format!("{remote_port}:127.0.0.1:{echo_port}")) .arg("--session") .arg("default") .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("local") .env("HOME", dir.path()) .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::piped()) .spawn() .unwrap(); let mut stream = connect_with_retry(remote_port, Duration::from_secs(5), &mut client); stream .set_read_timeout(Some(Duration::from_secs(3))) .unwrap(); stream.write_all(b"dosh-remote-forward-ping").unwrap(); let mut buf = [0u8; 64]; let n = stream.read(&mut buf).unwrap(); let _ = client.kill(); let client_output = client.wait_with_output().unwrap(); let _ = server.kill(); let _ = server.wait(); assert_eq!(&buf[..n], b"dosh-remote-forward-ping"); assert!( client_output.status.success() || client_output.status.code().is_none(), "stderr={}", String::from_utf8_lossy(&client_output.stderr) ); } #[test] fn native_dynamic_forward_socks_echo_smoke() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); write_native_client_auth(&dir, &config); let echo_port = start_echo_server(); let socks_port = free_tcp_port(); let mut server = start_server(&dir, &config); let client_bin = env!("CARGO_BIN_EXE_dosh-client"); let mut client = Command::new(client_bin) .arg("--auth") .arg("native") .arg("--no-cache") .arg("-N") .arg("-D") .arg(socks_port.to_string()) .arg("--session") .arg("default") .arg("--dosh-host") .arg("127.0.0.1") .arg("--dosh-port") .arg(port.to_string()) .arg("local") .env("HOME", dir.path()) .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::piped()) .spawn() .unwrap(); let mut stream = socks5_connect(socks_port, echo_port, &mut client); stream.write_all(b"dosh-dynamic-forward-ping").unwrap(); let mut buf = [0u8; 64]; let n = stream.read(&mut buf).unwrap(); let _ = client.kill(); let client_output = client.wait_with_output().unwrap(); let _ = server.kill(); let _ = server.wait(); assert_eq!(&buf[..n], b"dosh-dynamic-forward-ping"); assert!( client_output.status.success() || client_output.status.code().is_none(), "stderr={}", String::from_utf8_lossy(&client_output.stderr) ); } #[test] fn ticket_attach_after_server_restart_smoke() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let first = attach_once(&dir, port, true); let _ = server.kill(); let _ = server.wait(); assert!( first.status.success(), "stderr={}", String::from_utf8_lossy(&first.stderr) ); let mut server = start_server(&dir, &config); let second = attach_once(&dir, port, true); let _ = server.kill(); let _ = server.wait(); let stderr = String::from_utf8_lossy(&second.stderr); assert!(second.status.success(), "stderr={stderr}"); assert!( stderr.contains("udp_ticket_attach_ready"), "stderr={stderr}" ); } #[test] fn view_only_input_is_rejected_by_server() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (view_socket, view_bootstrap, view_ok) = direct_attach(&config, port, "view-only"); let input = Input { bytes: b"echo DOSH_VIEW_ONLY_BUG\n".to_vec(), }; let packet = protocol::encode_encrypted( PacketKind::Input, view_ok.client_id, 2, 0, &view_bootstrap.session_key, CLIENT_TO_SERVER, &protocol::to_body(&input).unwrap(), ) .unwrap(); view_socket .send_to(&packet, format!("127.0.0.1:{port}")) .unwrap(); thread::sleep(Duration::from_millis(250)); let (_rw_socket, _rw_bootstrap, rw_ok) = direct_attach(&config, port, "read-write"); let snapshot = String::from_utf8_lossy(&rw_ok.snapshot); let _ = server.kill(); let _ = server.wait(); assert!( !snapshot.contains("DOSH_VIEW_ONLY_BUG"), "view-only input reached PTY; snapshot={snapshot:?}" ); } #[test] fn multiple_clients_share_one_session_screen() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (writer_socket, writer_bootstrap, writer_ok) = direct_attach(&config, port, "read-write"); let (reader_socket, reader_bootstrap, _reader_ok) = direct_attach(&config, port, "read-write"); let input = Input { bytes: b"printf DOSH_MULTI_CLIENT\\n\n".to_vec(), }; send_encrypted( &writer_socket, port, PacketKind::Input, writer_ok.client_id, 2, 0, &writer_bootstrap.session_key, &protocol::to_body(&input).unwrap(), ); let text = collect_frame_text(&reader_socket, &reader_bootstrap.session_key, 2000); let _ = server.kill(); let _ = server.wait(); assert!( text.contains("DOSH_MULTI_CLIENT"), "expected second client to see first client's output, got {text:?}" ); } #[test] fn server_retransmits_unacked_output_frames() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let input = Input { bytes: b"printf DOSH_RETRANSMIT\\n\n".to_vec(), }; send_encrypted( &socket, port, PacketKind::Input, ok.client_id, 2, 0, &bootstrap.session_key, &protocol::to_body(&input).unwrap(), ); let mut seen = Vec::new(); let deadline = std::time::Instant::now() + Duration::from_secs(4); while std::time::Instant::now() < deadline { if let Some((header, frame)) = recv_frame(&socket, &bootstrap.session_key) { if String::from_utf8_lossy(&frame.bytes).contains("DOSH_RETRANSMIT") { seen.push((header.seq, frame.output_seq)); } } } let _ = server.kill(); let _ = server.wait(); let max_same_output_seq = seen .iter() .map(|(_, output_seq)| { seen.iter() .filter(|(_, candidate)| candidate == output_seq) .count() }) .max() .unwrap_or(0); assert!( max_same_output_seq >= 10, "expected retransmission to continue past the old 8-attempt ceiling, seen={seen:?}" ); } #[test] fn resize_updates_pty_size() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let resize = Resize { cols: 100, rows: 10, }; send_encrypted( &socket, port, PacketKind::Resize, ok.client_id, 2, 0, &bootstrap.session_key, &protocol::to_body(&resize).unwrap(), ); thread::sleep(Duration::from_millis(100)); let input = Input { bytes: b"stty size\n".to_vec(), }; send_encrypted( &socket, port, PacketKind::Input, ok.client_id, 3, 0, &bootstrap.session_key, &protocol::to_body(&input).unwrap(), ); let text = collect_frame_text(&socket, &bootstrap.session_key, 2000); let _ = server.kill(); let _ = server.wait(); assert!( text.contains("10 100"), "expected resized pty to report 10 100, got {text:?}" ); } #[test] fn pty_exit_sends_closed_frame() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let input = Input { bytes: b"exit\n".to_vec(), }; send_encrypted( &socket, port, PacketKind::Input, ok.client_id, 2, 0, &bootstrap.session_key, &protocol::to_body(&input).unwrap(), ); let mut closed = false; let deadline = std::time::Instant::now() + Duration::from_secs(3); while std::time::Instant::now() < deadline { if let Some((_header, frame)) = recv_frame(&socket, &bootstrap.session_key) { if frame.closed { closed = true; break; } } } let _ = server.kill(); let _ = server.wait(); assert!(closed, "expected closed frame after shell exit"); } #[test] fn pty_sets_terminal_environment() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let input = Input { bytes: b"printf \"$TERM\\n\"\n".to_vec(), }; send_encrypted( &socket, port, PacketKind::Input, ok.client_id, 2, 0, &bootstrap.session_key, &protocol::to_body(&input).unwrap(), ); let text = collect_frame_text(&socket, &bootstrap.session_key, 2000); let _ = server.kill(); let _ = server.wait(); assert!( text.contains("xterm-256color"), "expected TERM=xterm-256color, got {text:?}" ); } #[test] fn live_output_forwards_terminal_control_sequences() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let input = Input { bytes: b"printf '\\033[?1049hDOSH_ALT_SCREEN\\033[?1049l\\n'\n".to_vec(), }; send_encrypted( &socket, port, PacketKind::Input, ok.client_id, 2, 0, &bootstrap.session_key, &protocol::to_body(&input).unwrap(), ); let text = collect_frame_text(&socket, &bootstrap.session_key, 2000); let _ = server.kill(); let _ = server.wait(); assert!( text.contains("\x1b[?1049h") && text.contains("\x1b[?1049l"), "expected raw alternate-screen control sequences, got {text:?}" ); } #[test] fn tui_control_sequences_survive_transport_verbatim() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let sequences = concat!( "\x1b[?1049h", // alternate screen on "\x1b[?2004h", // bracketed paste on "\x1b[?1000h", // mouse tracking on "\x1b[?1006h", // SGR mouse encoding on "\x1b]0;DOSH_TITLE\x07", // OSC title "\x1b[?25l", // cursor hidden "\x1b[12;34H", // absolute cursor movement "\x1b[1;31m", // bold red "DOSH_TUI_VERBATIM", "\x1b[0m", "\x1b[?25h", "\x1b[?1006l", "\x1b[?1000l", "\x1b[?2004l", "\x1b[?1049l" ); let input = Input { bytes: format!("printf '{sequences}'\n").into_bytes(), }; send_encrypted( &socket, port, PacketKind::Input, ok.client_id, 2, 0, &bootstrap.session_key, &protocol::to_body(&input).unwrap(), ); let text = collect_frame_text(&socket, &bootstrap.session_key, 2000); let _ = server.kill(); let _ = server.wait(); for needle in [ "\x1b[?1049h", "\x1b[?2004h", "\x1b[?1000h", "\x1b[?1006h", "\x1b]0;DOSH_TITLE\x07", "\x1b[?25l", "\x1b[12;34H", "\x1b[1;31m", "DOSH_TUI_VERBATIM", "\x1b[0m", "\x1b[?25h", "\x1b[?1006l", "\x1b[?1000l", "\x1b[?2004l", "\x1b[?1049l", ] { assert!( text.contains(needle), "expected TUI control sequence {needle:?} to survive verbatim, got {text:?}" ); } } #[test] fn unicode_output_survives_transport() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let marker = "DOSH_UNICODE 你 e\u{301} 🚀"; let input = Input { bytes: format!("printf '{}\\n'\n", marker).into_bytes(), }; send_encrypted( &socket, port, PacketKind::Input, ok.client_id, 2, 0, &bootstrap.session_key, &protocol::to_body(&input).unwrap(), ); let text = collect_frame_text(&socket, &bootstrap.session_key, 2000); let _ = server.kill(); let _ = server.wait(); assert!( text.contains(marker), "expected UTF-8 output marker to survive transport, got {text:?}" ); } #[test] fn tui_graph_glyphs_survive_fast_repaints_without_snapshot_substitution() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config_with_output_interval(&dir, port, 1000); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let graph = "DOSH_GRAPH ⣀⣤⣶⣿ █▇▆▅▄▃▂▁ ╭╮╯╰"; let input = Input { bytes: format!( "printf '\\033[?1049h\\033[?25l'; for i in 1 2 3 4 5; do printf '\\033[H{} %s\\n' \"$i\"; done; printf '\\033[?25h\\033[?1049l'\n", graph ) .into_bytes(), }; send_encrypted( &socket, port, PacketKind::Input, ok.client_id, 2, 0, &bootstrap.session_key, &protocol::to_body(&input).unwrap(), ); let text = collect_frame_text(&socket, &bootstrap.session_key, 3000); let _ = server.kill(); let _ = server.wait(); assert!( text.contains(graph) && text.matches("DOSH_GRAPH").count() >= 5, "expected repeated raw graph glyph frames, got {text:?}" ); assert!( text.contains("\x1b[?25l") && text.contains("\x1b[?25h"), "expected cursor visibility controls to survive around graph repaint, got {text:?}" ); } #[test] fn btop_style_graph_output_stays_raw_when_retransmit_history_overflows() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config_with_retransmit_window(&dir, port, 3); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let graph = "DOSH_BTOP_NET ⠀⠁⠃⠇⡇⣇⣧⣷⣿"; let input = Input { bytes: format!( "stty -echo; \ printf '\\033[?1049h\\033[?2026h\\033[?25l'; \ for i in 1 2 3 4 5 6 7 8 9 10; do \ printf '\\033[6;4H\\033[38;2;80;220;140m{} %s\\033[0m' \"$i\"; \ done; \ printf '\\033[?25h\\033[?2026l\\033[?1049l'\n", graph ) .into_bytes(), }; send_encrypted( &socket, port, PacketKind::Input, ok.client_id, 2, 0, &bootstrap.session_key, &protocol::to_body(&input).unwrap(), ); let mut text = String::new(); let mut snapshots = 0usize; let deadline = std::time::Instant::now() + Duration::from_secs(3); while std::time::Instant::now() < deadline && !(text.matches("DOSH_BTOP_NET").count() >= 10 && text.contains("\x1b[?2026l")) { if let Some((_header, frame)) = recv_frame(&socket, &bootstrap.session_key) { if frame.snapshot { snapshots += 1; } text.push_str(&String::from_utf8_lossy(&frame.bytes)); } } let _ = server.kill(); let _ = server.wait(); assert_eq!( snapshots, 0, "live btop-style graph output must not be replaced by snapshots; got {snapshots}" ); assert!( text.matches("DOSH_BTOP_NET").count() >= 10 && text.contains("⣿") && text.contains("\x1b[38;2;80;220;140m") && text.contains("\x1b[?2026h") && text.contains("\x1b[?2026l"), "expected raw btop-style graph output to survive under retransmit pressure, got {text:?}" ); } #[test] fn large_tui_paint_is_delivered_in_mtu_safe_frames() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let input = Input { bytes: b"printf '\\033[?1049h'; yes DOSH_TUI_BIG_PAINT | head -n 300 | tr -d '\\n'; printf '\\033[?1049l'\n".to_vec(), }; send_encrypted( &socket, port, PacketKind::Input, ok.client_id, 2, 0, &bootstrap.session_key, &protocol::to_body(&input).unwrap(), ); let mut text = String::new(); let mut saw_split_frame = false; let deadline = std::time::Instant::now() + Duration::from_secs(3); while std::time::Instant::now() < deadline && text.matches("DOSH_TUI_BIG_PAINT").count() < 200 { if let Some((_header, frame)) = recv_frame(&socket, &bootstrap.session_key) { if frame.bytes.len() <= 1400 && frame.bytes.len() >= 900 { saw_split_frame = true; } text.push_str(&String::from_utf8_lossy(&frame.bytes)); } } let _ = server.kill(); let _ = server.wait(); assert!( text.contains("\x1b[?1049h") && text.matches("DOSH_TUI_BIG_PAINT").count() >= 200, "large alternate-screen paint did not survive transport, enter_alt={} markers={} bytes={}", text.contains("\x1b[?1049h"), text.matches("DOSH_TUI_BIG_PAINT").count(), text.len() ); assert!( saw_split_frame, "expected large TUI paint to be split into MTU-safe frames" ); } #[test] fn resume_snapshot_preserves_alternate_screen_mode() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let input = Input { bytes: b"printf '\\033[?1049hALT_SNAPSHOT'\n".to_vec(), }; send_encrypted( &socket, port, PacketKind::Input, ok.client_id, 2, 0, &bootstrap.session_key, &protocol::to_body(&input).unwrap(), ); let _ = collect_frame_text(&socket, &bootstrap.session_key, 500); let new_socket = UdpSocket::bind("127.0.0.1:0").unwrap(); new_socket .set_read_timeout(Some(Duration::from_secs(2))) .unwrap(); let resume = ResumeRequest { session: "default".to_string(), last_rendered_seq: ok.initial_seq, cols: 80, rows: 24, }; send_encrypted( &new_socket, port, PacketKind::ResumeRequest, ok.client_id, 3, 0, &bootstrap.session_key, &protocol::to_body(&resume).unwrap(), ); let (_header, resume_frame) = recv_frame(&new_socket, &bootstrap.session_key).expect("resume response"); let _ = server.kill(); let _ = server.wait(); assert!(resume_frame.snapshot); assert!( resume_frame.bytes.starts_with(b"\x1b[?1049h\x1b[H\x1b[2J"), "snapshot did not enter and clear alternate screen first: {:?}", String::from_utf8_lossy(&resume_frame.bytes) ); } #[test] fn resume_updates_udp_endpoint_for_roaming() { let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (_old_socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let new_socket = UdpSocket::bind("127.0.0.1:0").unwrap(); new_socket .set_read_timeout(Some(Duration::from_secs(2))) .unwrap(); let resume = ResumeRequest { session: "default".to_string(), last_rendered_seq: ok.initial_seq, cols: 80, rows: 24, }; send_encrypted( &new_socket, port, PacketKind::ResumeRequest, ok.client_id, 1, 0, &bootstrap.session_key, &protocol::to_body(&resume).unwrap(), ); let (_header, resume_frame) = recv_frame(&new_socket, &bootstrap.session_key).expect("resume response on new socket"); assert!(resume_frame.snapshot); let input = Input { bytes: b"printf DOSH_ROAM\\n\n".to_vec(), }; send_encrypted( &new_socket, port, PacketKind::Input, ok.client_id, 2, resume_frame.output_seq, &bootstrap.session_key, &protocol::to_body(&input).unwrap(), ); let text = collect_frame_text(&new_socket, &bootstrap.session_key, 2000); let _ = server.kill(); let _ = server.wait(); assert!( text.contains("DOSH_ROAM"), "expected output on resumed socket, got {text:?}" ); } #[test] #[ignore = "30-minute launch soak; run with `DOSH_SOAK_SECONDS=1800 cargo test --test integration_smoke sleep_roaming_soak_30m -- --ignored --nocapture`"] fn sleep_roaming_soak_30m() { let soak_secs = std::env::var("DOSH_SOAK_SECONDS") .ok() .and_then(|value| value.parse::().ok()) .unwrap_or(30 * 60); let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config_with_timeout(&dir, port, soak_secs + 60); let mut server = start_server(&dir, &config); let (_old_socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); thread::sleep(Duration::from_secs(soak_secs)); let resumed = UdpSocket::bind("127.0.0.1:0").unwrap(); resumed .set_read_timeout(Some(Duration::from_secs(3))) .unwrap(); let resume = ResumeRequest { session: "default".to_string(), last_rendered_seq: ok.initial_seq, cols: 80, rows: 24, }; send_encrypted( &resumed, port, PacketKind::ResumeRequest, ok.client_id, 1, 0, &bootstrap.session_key, &protocol::to_body(&resume).unwrap(), ); let (_header, resume_frame) = recv_frame(&resumed, &bootstrap.session_key).expect("resume response after soak"); assert!(resume_frame.snapshot); send_encrypted( &resumed, port, PacketKind::Input, ok.client_id, 2, resume_frame.output_seq, &bootstrap.session_key, &protocol::to_body(&Input { bytes: b"printf DOSH_SOAK_OK\\n\n".to_vec(), }) .unwrap(), ); let text = collect_frame_text(&resumed, &bootstrap.session_key, 3000); let _ = server.kill(); let _ = server.wait(); assert!( text.contains("DOSH_SOAK_OK"), "session did not survive {soak_secs}s idle/roaming soak; output={text:?}" ); } #[test] fn transport_rekey_round_trip_keeps_session_alive() { use dosh::native::derive_rekey_session_key; use dosh::protocol::Rekey; let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); // Rotate after a single packet so a rekey fires almost immediately. let mut raw = fs::read_to_string(&config).unwrap(); raw.push_str("rekey_after_packets = 1\n"); fs::write(&config, raw).unwrap(); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); socket .set_read_timeout(Some(Duration::from_millis(300))) .unwrap(); // Track the live transport key; it rotates when a Rekey arrives. let mut current_key = bootstrap.session_key; let mut current_key_id = protocol::session_key_id(¤t_key); let mut send_seq = 2u64; // Produce some output so the server starts sending frames (and rekeys). let input = Input { bytes: b"printf DOSH_REKEY_ONE\\n\n".to_vec(), }; send_encrypted( &socket, port, PacketKind::Input, ok.client_id, send_seq, 0, ¤t_key, &protocol::to_body(&input).unwrap(), ); send_seq += 1; // Drive the loop: decrypt frames under the live key, and when a Rekey lands, // adopt the new epoch key and ack it. Confirm a post-rekey input still works. let mut rekeyed = false; let mut saw_post_rekey_output = false; let mut buf = [0u8; 65535]; let deadline = std::time::Instant::now() + Duration::from_secs(8); while std::time::Instant::now() < deadline { let Ok((n, _)) = socket.recv_from(&mut buf) else { continue; }; let Ok(packet) = protocol::decode(&buf[..n]) else { continue; }; match packet.header.kind { PacketKind::Rekey => { let plain = protocol::decrypt_body(&packet, ¤t_key, SERVER_TO_CLIENT).unwrap(); let rekey: Rekey = protocol::from_body(&plain).unwrap(); let new_key = derive_rekey_session_key( ¤t_key, &rekey.rekey_material, ¤t_key_id, rekey.epoch, ) .unwrap(); assert_eq!( protocol::session_key_id(&new_key), rekey.new_session_key_id, "client-derived rekey key id must match the server's" ); current_key = new_key; current_key_id = rekey.new_session_key_id; // Ack under the NEW key. send_encrypted( &socket, port, PacketKind::RekeyAck, ok.client_id, send_seq, 0, ¤t_key, b"", ); send_seq += 1; rekeyed = true; // Now send a fresh input under the new key. let input = Input { bytes: b"printf DOSH_REKEY_TWO\\n\n".to_vec(), }; send_encrypted( &socket, port, PacketKind::Input, ok.client_id, send_seq, 0, ¤t_key, &protocol::to_body(&input).unwrap(), ); send_seq += 1; } PacketKind::Frame | PacketKind::ResumeOk => { if let Ok(plain) = protocol::decrypt_body(&packet, ¤t_key, SERVER_TO_CLIENT) { if let Ok(frame) = protocol::from_body::(&plain) { let text = String::from_utf8_lossy(&frame.bytes); if rekeyed && text.contains("DOSH_REKEY_TWO") { saw_post_rekey_output = true; break; } } } } _ => {} } } let _ = server.kill(); let _ = server.wait(); assert!(rekeyed, "server never initiated a rekey"); assert!( saw_post_rekey_output, "post-rekey input/output round trip failed under the new epoch key" ); } #[test] fn input_from_new_source_address_migrates_connection() { // Spec §11: the server must accept client source-address migration after ANY // valid authenticated/encrypted packet from a new address, not just resume. let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); let mut server = start_server(&dir, &config); let (_old_socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); // A fresh socket = a new source address (new ephemeral port). The very first // packet from it is an ordinary encrypted Input, not a ResumeRequest. let new_socket = UdpSocket::bind("127.0.0.1:0").unwrap(); new_socket .set_read_timeout(Some(Duration::from_secs(2))) .unwrap(); let input = Input { bytes: b"printf DOSH_MIGRATE\\n\n".to_vec(), }; let packet = protocol::encode_encrypted( PacketKind::Input, ok.client_id, 2, 0, &bootstrap.session_key, CLIENT_TO_SERVER, &protocol::to_body(&input).unwrap(), ) .unwrap(); new_socket .send_to(&packet, format!("127.0.0.1:{port}")) .unwrap(); // Output frames for this input must now be delivered to the NEW socket, // proving the server migrated `endpoint` off the original address. let text = collect_frame_text(&new_socket, &bootstrap.session_key, 2000); let _ = server.kill(); let _ = server.wait(); assert!( text.contains("DOSH_MIGRATE"), "expected server output on the migrated socket, got {text:?}" ); } #[test] fn native_auth_rate_limit_rejects_flood_before_crypto() { use dosh::native::{NATIVE_PROTOCOL_VERSION, NativeClientHello}; use dosh::protocol::NativeClientHelloBody; let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); // Squeeze the rate limit down to 2/min so a short burst trips it. let mut raw = fs::read_to_string(&config).unwrap(); raw.push_str("native_auth_rate_limit_per_minute = 2\n"); fs::write(&config, raw).unwrap(); write_native_client_auth(&dir, &config); let mut server = start_server(&dir, &config); let socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); socket .set_read_timeout(Some(Duration::from_secs(2))) .unwrap(); let user = std::env::var("USER").unwrap_or_else(|_| "unknown".to_string()); let make_hello = || { let (_, client_public) = dosh::native::generate_native_ephemeral(); let hello = NativeClientHello { protocol_version: NATIVE_PROTOCOL_VERSION, client_random: crypto::random_32(), client_ephemeral_public: client_public, requested_host: "local".to_string(), requested_user: user.clone(), requested_session: "default".to_string(), requested_mode: "read-write".to_string(), terminal_size: (80, 24), supported_aead: vec!["chacha20poly1305".to_string()], supported_user_key_algorithms: vec!["ssh-ed25519".to_string()], cached_host_key_fingerprint: None, attach_ticket_envelope: None, requested_env: Vec::new(), }; protocol::encode_plain( PacketKind::NativeClientHello, [0u8; 16], 1, 0, &protocol::to_body(&NativeClientHelloBody { hello }).unwrap(), ) .unwrap() }; let mut reasons = Vec::new(); // Burst of hellos; with a 2/min budget the later ones must be rejected. for _ in 0..6 { socket .send_to(&make_hello(), format!("127.0.0.1:{port}")) .unwrap(); let mut buf = [0u8; 65535]; if let Ok((n, _)) = socket.recv_from(&mut buf) { let packet = protocol::decode(&buf[..n]).unwrap(); if packet.header.kind == PacketKind::AttachReject { let reject: AttachReject = protocol::from_body(&packet.body).unwrap(); reasons.push(reject.reason); } } } let _ = server.kill(); let _ = server.wait(); assert!( reasons.iter().any(|r| r.contains("rate limit")), "expected a rate-limit reject within the burst, got {reasons:?}" ); } #[test] fn native_hello_with_mismatched_protocol_version_gets_named_reject_not_hang() { use dosh::native::{NATIVE_PROTOCOL_VERSION, NativeClientHello}; use dosh::protocol::{NativeClientHelloBody, VERSION_MISMATCH_REASON}; let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config(&dir, port); write_native_client_auth(&dir, &config); let mut server = start_server(&dir, &config); let socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); socket .set_read_timeout(Some(Duration::from_secs(2))) .unwrap(); // Same wire VERSION as the server (so the datagram decodes), but an // incompatible native handshake protocol_version. This is the spec's // plaintext negotiation point: the server must answer with a clear, named // reject rather than letting the client time out. let hello = NativeClientHello { protocol_version: NATIVE_PROTOCOL_VERSION.wrapping_add(1), client_random: crypto::random_32(), client_ephemeral_public: [3u8; 32], requested_host: "local".to_string(), requested_user: std::env::var("USER").unwrap_or_else(|_| "unknown".to_string()), requested_session: "default".to_string(), requested_mode: "read-write".to_string(), terminal_size: (80, 24), supported_aead: vec!["chacha20poly1305".to_string()], supported_user_key_algorithms: vec!["ssh-ed25519".to_string()], cached_host_key_fingerprint: None, attach_ticket_envelope: None, requested_env: Vec::new(), }; let packet = protocol::encode_plain( PacketKind::NativeClientHello, [0u8; 16], 1, 0, &protocol::to_body(&NativeClientHelloBody { hello }).unwrap(), ) .unwrap(); socket .send_to(&packet, format!("127.0.0.1:{port}")) .unwrap(); let mut buf = [0u8; 65535]; // recv with the 2s read timeout above: a hang would surface as a recv error // here instead of a reject, which is exactly what this test guards against. let (n, _) = socket .recv_from(&mut buf) .expect("server must answer a version-mismatched hello, not hang"); let packet = protocol::decode(&buf[..n]).unwrap(); let reject: AttachReject = protocol::from_body(&packet.body).unwrap(); let _ = server.kill(); let _ = server.wait(); assert_eq!(packet.header.kind, PacketKind::AttachReject); assert!( reject.reason.contains(VERSION_MISMATCH_REASON), "expected a named protocol-version-mismatch reject, got {:?}", reject.reason ); } /// End-to-end SSH-agent forwarding: with -A opt-in AND allow_agent_forwarding on, /// the server exports a proxy SSH_AUTH_SOCK into the session and tunnels a /// connection from a "remote process" (here, the test connecting to the proxy /// socket) all the way back to the client's local ssh-agent. We drive an /// identities request through the chain and verify the real agent's answer comes /// back, proving server-proxy -> Dosh stream -> client unix splice -> local agent. #[test] fn native_agent_forwarding_round_trip() { use portable_pty::{CommandBuilder, NativePtySystem, PtySize, PtySystem}; let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config_with_agent_forwarding(&dir, port); write_native_client_auth(&dir, &config); // Local fake agent the client will splice forwarded connections into. let agent_sock = dir.path().join("local-agent.sock"); start_fake_ssh_agent_looping(&agent_sock, "forwarded-agent-key"); // Controlled TMPDIR so the server's proxy socket path is deterministic and // isolated to this test (std::env::temp_dir honors TMPDIR). let tmpdir = dir.path().join("tmp"); fs::create_dir_all(&tmpdir).unwrap(); let server_bin = env!("CARGO_BIN_EXE_dosh-server"); let mut server = Command::new(server_bin) .arg("serve") .arg("--config") .arg(&config) .env("HOME", dir.path()) .env("TMPDIR", &tmpdir) .stdout(Stdio::null()) .stderr(Stdio::null()) .spawn() .unwrap(); thread::sleep(Duration::from_millis(500)); // Spawn the client inside a real PTY so it can enter raw mode and stay // attached to an interactive session (agent forwarding needs a spawned shell). let pty = NativePtySystem::default(); let pair = pty .openpty(PtySize { rows: 24, cols: 80, pixel_width: 0, pixel_height: 0, }) .unwrap(); let client_bin = env!("CARGO_BIN_EXE_dosh-client"); let mut cmd = CommandBuilder::new(client_bin); cmd.args([ "--auth", "native", "--no-cache", "-A", "--session", "agent-session", "--dosh-host", "127.0.0.1", "--dosh-port", &port.to_string(), "local", ]); cmd.env("HOME", dir.path().to_string_lossy().to_string()); cmd.env("TMPDIR", tmpdir.to_string_lossy().to_string()); cmd.env("SSH_AUTH_SOCK", agent_sock.to_string_lossy().to_string()); let mut child = pair.slave.spawn_command(cmd).unwrap(); drop(pair.slave); // The server's proxy socket is created at session spawn; its path is // deterministic: TMPDIR/dosh-agent-/agent..0.sock. let server_pid = server.id(); let proxy_sock = tmpdir .join(format!("dosh-agent-{server_pid}")) .join(format!("agent.{server_pid}.0.sock")); // Wait for the client to authenticate and the server to spawn the shell + // bind the proxy socket, then connect as the "remote process" and run an // identities request through the forwarded chain. let deadline = std::time::Instant::now() + Duration::from_secs(10); let answer = loop { if std::time::Instant::now() >= deadline { let _ = child.kill(); let _ = server.kill(); let _ = server.wait(); panic!("agent proxy socket {proxy_sock:?} never became usable"); } if proxy_sock.exists() && let Ok(mut stream) = UnixStream::connect(&proxy_sock) { stream .set_read_timeout(Some(Duration::from_secs(3))) .unwrap(); if write_agent_packet(&mut stream, &[11]).is_ok() && let Ok(answer) = read_agent_packet(&mut stream) { break answer; } } thread::sleep(Duration::from_millis(100)); }; let _ = child.kill(); let _ = child.wait(); let _ = server.kill(); let _ = server.wait(); // The answer must be an IDENTITIES_ANSWER (12) carrying our agent's comment, // proving the bytes traversed the full forwarding chain to the local agent. assert_eq!( answer.first(), Some(&12), "expected IDENTITIES_ANSWER from the forwarded agent, got {:?}", answer.first() ); assert!( answer .windows(b"forwarded-agent-key".len()) .any(|w| w == b"forwarded-agent-key"), "forwarded agent answer should carry the local agent's key comment" ); } /// Agent forwarding stays off unless the client opts in: with the server allowing /// it but no -A, no proxy socket is created. #[test] fn native_agent_forwarding_requires_client_opt_in() { use portable_pty::{CommandBuilder, NativePtySystem, PtySize, PtySystem}; let dir = tempfile::tempdir().unwrap(); let port = free_udp_port(); let config = write_server_config_with_agent_forwarding(&dir, port); write_native_client_auth(&dir, &config); let agent_sock = dir.path().join("local-agent.sock"); start_fake_ssh_agent_looping(&agent_sock, "forwarded-agent-key"); let tmpdir = dir.path().join("tmp"); fs::create_dir_all(&tmpdir).unwrap(); let server_bin = env!("CARGO_BIN_EXE_dosh-server"); let mut server = Command::new(server_bin) .arg("serve") .arg("--config") .arg(&config) .env("HOME", dir.path()) .env("TMPDIR", &tmpdir) .stdout(Stdio::null()) .stderr(Stdio::null()) .spawn() .unwrap(); thread::sleep(Duration::from_millis(500)); let pty = NativePtySystem::default(); let pair = pty .openpty(PtySize { rows: 24, cols: 80, pixel_width: 0, pixel_height: 0, }) .unwrap(); let client_bin = env!("CARGO_BIN_EXE_dosh-client"); let mut cmd = CommandBuilder::new(client_bin); // Note: NO -A flag. cmd.args([ "--auth", "native", "--no-cache", "--session", "no-agent-session", "--dosh-host", "127.0.0.1", "--dosh-port", &port.to_string(), "local", ]); cmd.env("HOME", dir.path().to_string_lossy().to_string()); cmd.env("TMPDIR", tmpdir.to_string_lossy().to_string()); cmd.env("SSH_AUTH_SOCK", agent_sock.to_string_lossy().to_string()); let mut child = pair.slave.spawn_command(cmd).unwrap(); drop(pair.slave); // Give the client time to authenticate and the shell to spawn. thread::sleep(Duration::from_secs(2)); let server_pid = server.id(); let agent_dir = tmpdir.join(format!("dosh-agent-{server_pid}")); let _ = child.kill(); let _ = child.wait(); let _ = server.kill(); let _ = server.wait(); // No client opt-in -> the server must not have created any agent socket dir. assert!( !agent_dir.exists(), "agent proxy dir {agent_dir:?} must not exist without client -A opt-in" ); } // --------------------------------------------------------------------------- // Session persistence across a server restart. // --------------------------------------------------------------------------- /// Write a server config with `persist_sessions = true` so each session's shell /// runs in a detached holder that survives a server restart. fn write_persistent_server_config(dir: &tempfile::TempDir, port: u16) -> std::path::PathBuf { let config = write_server_config(dir, port); let mut raw = fs::read_to_string(&config).unwrap(); // The base writer hardcodes `persist_sessions = false`; flip it on. raw = raw.replace("persist_sessions = false", "persist_sessions = true"); // Persistence tests attach the sessions they need explicitly. Leaving the // base prewarm on here creates a detached default holder in every test. raw = raw.replace("prewarm_sessions = [\"default\"]", "prewarm_sessions = []"); fs::write(&config, raw).unwrap(); config } /// Mirror the server's hex encoding of a session name to its runtime dir. fn session_runtime_dir(sessions_dir: &Path, session: &str) -> PathBuf { let mut hex = String::new(); for b in session.as_bytes() { hex.push_str(&format!("{b:02x}")); } sessions_dir.join("run").join(hex) } /// Read a holder's recorded shell pid from its meta file, if present. fn holder_shell_pid(sessions_dir: &Path, session: &str) -> Option { let meta = session_runtime_dir(sessions_dir, session).join("meta"); let body = fs::read_to_string(meta).ok()?; body.lines().nth(1)?.parse().ok() } /// Tear down a persistent session's holder + shell so a test never leaks them /// (the TempDir drop would otherwise leave detached processes running). fn kill_holder(sessions_dir: &Path, session: &str) { if let Some(pid) = holder_shell_pid(sessions_dir, session) { let _ = Command::new("kill").arg("-9").arg(pid.to_string()).status(); } // Best-effort: also kill any matching holder process for this runtime dir. let dir = session_runtime_dir(sessions_dir, session); let _ = Command::new("pkill") .arg("-9") .arg("-f") .arg(format!("dosh-server hold --runtime-dir {}", dir.display())) .status(); } /// Create a holder the way older Dosh builds did for implicit sessions, so /// startup migration can be tested without depending on the new spawn policy. fn spawn_legacy_holder(sessions_dir: &Path, session: &str) { let server = env!("CARGO_BIN_EXE_dosh-server"); let runtime_dir = dosh::persist::ensure_runtime_dir(sessions_dir, session).unwrap(); let mut launcher = Command::new(server) .arg("hold") .arg("--runtime-dir") .arg(&runtime_dir) .arg("--session") .arg(session) .arg("--shell") .arg("/bin/sh") .arg("--cols") .arg("80") .arg("--rows") .arg("24") .stdout(Stdio::null()) .stderr(Stdio::null()) .spawn() .unwrap(); let deadline = std::time::Instant::now() + Duration::from_secs(5); while std::time::Instant::now() < deadline { if holder_shell_pid(sessions_dir, session).is_some() && runtime_dir.join("holder.sock").exists() { let _ = launcher.wait(); return; } if let Some(status) = launcher.try_wait().unwrap() { panic!("legacy holder launcher exited before ready: {status}"); } thread::sleep(Duration::from_millis(20)); } let _ = launcher.kill(); let _ = launcher.wait(); panic!("legacy holder did not become ready"); } /// Send one encrypted Input line and give the shell a moment to act. fn type_line(socket: &UdpSocket, port: u16, ok: &AttachOk, key: &[u8; 32], seq: u64, line: &str) { let input = Input { bytes: line.as_bytes().to_vec(), }; send_encrypted( socket, port, PacketKind::Input, ok.client_id, seq, 0, key, &protocol::to_body(&input).unwrap(), ); thread::sleep(Duration::from_millis(150)); } /// Proof of survival: a session's shell + state + screen outlive a full /// `dosh-server` process restart. /// /// We attach, set durable shell state (`cd /tmp`, `export MARK=...`) and paint a /// recognizable marker on the screen, wait for the screen to be mirrored to disk, /// then KILL the server process (simulating a crash/upgrade). We restart the /// server on the same config/sockets, reattach, and assert we land on the SAME /// shell — the exported variable and working directory are still there, and the /// restored attach snapshot still shows the pre-restart screen — not a fresh /// shell. #[test] fn session_survives_server_restart_same_shell_and_screen() { let dir = tempfile::tempdir().unwrap(); let sessions_dir = dir.path().join("sessions"); let port = free_udp_port(); let config = write_persistent_server_config(&dir, port); let mut server = start_server(&dir, &config); // First attach: establish durable shell state and a screen marker. let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); let key = bootstrap.session_key; type_line(&socket, port, &ok, &key, 2, "cd /tmp\n"); type_line( &socket, port, &ok, &key, 3, "export MARK=zebra_persist_42\n", ); // Paint a unique, stable marker into the screen (printf, no trailing newline // scroll surprises) so the restored snapshot is easy to recognize. type_line( &socket, port, &ok, &key, 4, "printf 'PRE_RESTART_SCREEN_MARKER\\n'\n", ); // Drain output so the parser/screen state and disk mirror catch up. let pre = collect_frame_text(&socket, &key, 1500); assert!( pre.contains("PRE_RESTART_SCREEN_MARKER"), "expected to see the screen marker before restart, got {pre:?}" ); // The periodic flush mirrors the screen every ~2s; wait long enough that the // pre-restart screen is guaranteed on disk before we crash the server. thread::sleep(Duration::from_secs(3)); let shell_pid_before = holder_shell_pid(&sessions_dir, "default"); assert!( shell_pid_before.is_some(), "a holder shell pid should be recorded for a persistent session" ); // CRASH: kill the server process outright (its in-memory state is lost; only // the holder + shell + on-disk screen remain). let _ = server.kill(); let _ = server.wait(); thread::sleep(Duration::from_millis(300)); // The holder + shell must still be alive after the server died. let pid = shell_pid_before.unwrap(); let alive = unsafe { libc::kill(pid, 0) } == 0; assert!( alive, "shell pid {pid} must survive the server crash (holder keeps it alive)" ); // RESTART on the same config/sockets; the server re-adopts the holder. let mut server = start_server(&dir, &config); // Reattach. A fresh bootstrap attach lands on the EXISTING re-adopted // session (same name), not a new shell. let (socket2, bootstrap2, ok2) = direct_attach(&config, port, "read-write"); let key2 = bootstrap2.session_key; // 1) Screen survived: the restored attach snapshot still shows the marker. let snapshot = String::from_utf8_lossy(&ok2.snapshot).to_string(); // 2) Shell identity + state survived: ask the SAME shell for its state. type_line( &socket2, port, &ok2, &key2, 2, "printf 'MARK=%s PWD=%s\\n' \"$MARK\" \"$PWD\"\n", ); let post = collect_frame_text(&socket2, &key2, 2000); let shell_pid_after = holder_shell_pid(&sessions_dir, "default"); // Clean up BEFORE asserting so a failed assert never leaks the holder. let _ = server.kill(); let _ = server.wait(); kill_holder(&sessions_dir, "default"); assert_eq!( shell_pid_after, shell_pid_before, "re-adopted session must be backed by the same shell pid, not a fresh one" ); assert!( post.contains("MARK=zebra_persist_42"), "exported variable must survive the restart (same shell), got {post:?}" ); assert!( post.contains("PWD=/tmp"), "working directory must survive the restart (same shell), got {post:?}" ); assert!( snapshot.contains("PRE_RESTART_SCREEN_MARKER"), "restored attach snapshot must repaint the pre-restart screen, got {snapshot:?}" ); } #[test] fn implicit_session_survives_update_restart_via_ticket_reattach() { let dir = tempfile::tempdir().unwrap(); let sessions_dir = dir.path().join("sessions"); let port = free_udp_port(); let config = write_persistent_server_config(&dir, port); let session = protocol::generate_implicit_session_name(); let mut server = start_server(&dir, &config); let (socket, bootstrap, ok) = direct_attach_session(&config, port, "read-write", &session); let key = bootstrap.session_key; type_line(&socket, port, &ok, &key, 2, "cd /tmp\n"); type_line( &socket, port, &ok, &key, 3, "export IMPLICIT_MARK=implicit_restart_42\n", ); type_line( &socket, port, &ok, &key, 4, "printf 'IMPLICIT_SCREEN_MARKER\\n'\n", ); let pre = collect_frame_text(&socket, &key, 1500); assert!( pre.contains("IMPLICIT_SCREEN_MARKER"), "implicit marker missing before restart: {pre:?}" ); thread::sleep(Duration::from_secs(3)); let shell_pid_before = holder_shell_pid(&sessions_dir, &session); assert!( shell_pid_before.is_some(), "implicit sessions must get temporary holders so active clients survive updates" ); let _ = server.kill(); let _ = server.wait(); thread::sleep(Duration::from_millis(300)); let mut server = start_server(&dir, &config); let resume = ResumeRequest { session: session.clone(), last_rendered_seq: ok.initial_seq, cols: 80, rows: 24, }; send_encrypted( &socket, port, PacketKind::ResumeRequest, ok.client_id, 5, 0, &key, &protocol::to_body(&resume).unwrap(), ); let mut buf = [0u8; 65535]; let deadline = std::time::Instant::now() + Duration::from_secs(2); loop { assert!( std::time::Instant::now() < deadline, "old live resume did not receive unknown-client reject" ); let (n, _) = socket.recv_from(&mut buf).unwrap(); let packet = protocol::decode(&buf[..n]).unwrap(); if packet.header.kind == PacketKind::Frame { continue; } assert_eq!(packet.header.kind, PacketKind::AttachReject); assert_eq!(packet.header.conn_id, ok.client_id); break; } let ok2 = ticket_attach_session(&socket, port, &bootstrap, &session, "read-write"); let key2 = ok2.session_key; let snapshot = String::from_utf8_lossy(&ok2.snapshot).to_string(); type_line( &socket, port, &ok2, &key2, 2, "printf 'IMPLICIT_MARK=%s PWD=%s\\n' \"$IMPLICIT_MARK\" \"$PWD\"\n", ); let post = collect_frame_text(&socket, &key2, 2000); let shell_pid_after = holder_shell_pid(&sessions_dir, &session); let _ = server.kill(); let _ = server.wait(); kill_holder(&sessions_dir, &session); assert!( snapshot.contains("IMPLICIT_SCREEN_MARKER"), "ticket reattach snapshot must repaint the exact pre-update screen, got {snapshot:?}" ); assert_eq!( shell_pid_after, shell_pid_before, "implicit update reconnect must use the same holder shell" ); assert!( post.contains("IMPLICIT_MARK=implicit_restart_42"), "implicit session env must survive update reconnect, got {post:?}" ); assert!( post.contains("PWD=/tmp"), "implicit session cwd must survive update reconnect, got {post:?}" ); } #[test] fn startup_readopts_implicit_holders_for_update_reconnect() { let dir = tempfile::tempdir().unwrap(); let sessions_dir = dir.path().join("sessions"); let port = free_udp_port(); let config = write_persistent_server_config(&dir, port); let session = protocol::generate_implicit_session_name(); spawn_legacy_holder(&sessions_dir, &session); let shell_pid = holder_shell_pid(&sessions_dir, &session).expect("legacy holder shell pid"); assert!( unsafe { libc::kill(shell_pid, 0) } == 0, "legacy holder shell must be live before startup" ); let mut server = start_server(&dir, &config); thread::sleep(Duration::from_millis(300)); let _ = server.kill(); let _ = server.wait(); assert!( holder_shell_pid(&sessions_dir, &session).is_some(), "startup must keep live implicit holders so active update reconnects can reattach" ); assert!( unsafe { libc::kill(shell_pid, 0) } == 0, "startup must not shut down a live implicit holder before reconnect grace" ); kill_holder(&sessions_dir, &session); } #[test] fn multiple_persistent_named_sessions_survive_restart_independently() { let dir = tempfile::tempdir().unwrap(); let sessions_dir = dir.path().join("sessions"); let port = free_udp_port(); let config = write_persistent_server_config(&dir, port); let mut server = start_server(&dir, &config); let (work_socket, work_bootstrap, work_ok) = direct_attach_session(&config, port, "read-write", "work"); let (logs_socket, logs_bootstrap, logs_ok) = direct_attach_session(&config, port, "read-write", "logs"); let work_key = work_bootstrap.session_key; let logs_key = logs_bootstrap.session_key; type_line( &work_socket, port, &work_ok, &work_key, 2, "export DOSH_SLOT=work_slot\n", ); type_line( &logs_socket, port, &logs_ok, &logs_key, 2, "export DOSH_SLOT=logs_slot\n", ); type_line( &work_socket, port, &work_ok, &work_key, 3, "printf 'WORK_SCREEN_MARKER\\n'\n", ); type_line( &logs_socket, port, &logs_ok, &logs_key, 3, "printf 'LOGS_SCREEN_MARKER\\n'\n", ); let work_pre = collect_frame_text(&work_socket, &work_key, 1500); let logs_pre = collect_frame_text(&logs_socket, &logs_key, 1500); assert!( work_pre.contains("WORK_SCREEN_MARKER"), "work marker missing before restart: {work_pre:?}" ); assert!( logs_pre.contains("LOGS_SCREEN_MARKER"), "logs marker missing before restart: {logs_pre:?}" ); thread::sleep(Duration::from_secs(3)); let work_pid_before = holder_shell_pid(&sessions_dir, "work"); let logs_pid_before = holder_shell_pid(&sessions_dir, "logs"); assert!(work_pid_before.is_some(), "work holder pid missing"); assert!(logs_pid_before.is_some(), "logs holder pid missing"); assert_ne!( work_pid_before, logs_pid_before, "named sessions must have independent holder shells" ); let _ = server.kill(); let _ = server.wait(); thread::sleep(Duration::from_millis(300)); let mut server = start_server(&dir, &config); let (work_socket2, work_bootstrap2, work_ok2) = direct_attach_session(&config, port, "read-write", "work"); let (logs_socket2, logs_bootstrap2, logs_ok2) = direct_attach_session(&config, port, "read-write", "logs"); let work_key2 = work_bootstrap2.session_key; let logs_key2 = logs_bootstrap2.session_key; let work_snapshot = String::from_utf8_lossy(&work_ok2.snapshot).to_string(); let logs_snapshot = String::from_utf8_lossy(&logs_ok2.snapshot).to_string(); type_line( &work_socket2, port, &work_ok2, &work_key2, 2, "printf 'WORK_SLOT=%s\\n' \"$DOSH_SLOT\"\n", ); type_line( &logs_socket2, port, &logs_ok2, &logs_key2, 2, "printf 'LOGS_SLOT=%s\\n' \"$DOSH_SLOT\"\n", ); let work_post = collect_frame_text(&work_socket2, &work_key2, 2000); let logs_post = collect_frame_text(&logs_socket2, &logs_key2, 2000); let work_pid_after = holder_shell_pid(&sessions_dir, "work"); let logs_pid_after = holder_shell_pid(&sessions_dir, "logs"); let _ = server.kill(); let _ = server.wait(); kill_holder(&sessions_dir, "work"); kill_holder(&sessions_dir, "logs"); assert_eq!(work_pid_after, work_pid_before); assert_eq!(logs_pid_after, logs_pid_before); assert!( work_snapshot.contains("WORK_SCREEN_MARKER") && !work_snapshot.contains("LOGS_SCREEN_MARKER"), "work snapshot should restore only work screen, got {work_snapshot:?}" ); assert!( logs_snapshot.contains("LOGS_SCREEN_MARKER") && !logs_snapshot.contains("WORK_SCREEN_MARKER"), "logs snapshot should restore only logs screen, got {logs_snapshot:?}" ); assert!( work_post.contains("WORK_SLOT=work_slot") && !work_post.contains("logs_slot"), "work state crossed sessions or disappeared: {work_post:?}" ); assert!( logs_post.contains("LOGS_SLOT=logs_slot") && !logs_post.contains("work_slot"), "logs state crossed sessions or disappeared: {logs_post:?}" ); }