dosh/tests/integration_smoke.rs

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::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,
};
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_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}"]
"#,
            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 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 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<u8>, 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();
}

fn read_agent_packet(stream: &mut UnixStream) -> std::io::Result<Vec<u8>> {
    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<u8>, 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) {
    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(),
        "default".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 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 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] forwarding policy"), "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_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 mut duplicate = None;
    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 String::from_utf8_lossy(&frame.bytes).contains("DOSH_RETRANSMIT") {
                if seen
                    .iter()
                    .any(|(_, output_seq)| *output_seq == frame.output_seq)
                {
                    duplicate = Some((header.seq, frame.output_seq));
                    break;
                }
                seen.push((header.seq, frame.output_seq));
            }
        }
    }

    let _ = server.kill();
    let _ = server.wait();

    assert!(
        duplicate.is_some(),
        "expected retransmitted same output seq, 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 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"),
        "snapshot did not enter 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]
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(&current_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,
        &current_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, &current_key, SERVER_TO_CLIENT).unwrap();
                let rekey: Rekey = protocol::from_body(&plain).unwrap();
                let new_key = derive_rekey_session_key(
                    &current_key,
                    &rekey.rekey_material,
                    &current_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,
                    &current_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,
                    &current_key,
                    &protocol::to_body(&input).unwrap(),
                );
                send_seq += 1;
            }
            PacketKind::Frame | PacketKind::ResumeOk => {
                if let Ok(plain) = protocol::decrypt_body(&packet, &current_key, SERVER_TO_CLIENT) {
                    if let Ok(frame) = protocol::from_body::<Frame>(&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
    );
}