Add launch hardening gates

tests/hostile_network.rs

@@ -20,7 +20,8 @@
 //! tests are reproducible and fast.
 
 use std::fs;
-use std::net::{SocketAddr, UdpSocket};
+use std::io::Read;
+use std::net::{SocketAddr, TcpListener, UdpSocket};
 use std::process::{Child, Command, Stdio};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
@@ -28,13 +29,18 @@ use std::sync::mpsc::{Receiver, Sender, channel};
 use std::thread;
 use std::time::{Duration, Instant};
 
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD;
 use dosh::auth::{BootstrapResponse, build_bootstrap, load_or_create_server_secret};
 use dosh::config::load_server_config;
 use dosh::crypto;
+use dosh::native::{self, ForwardingKind, ForwardingRequest, NativeAuthOk, NativeClientHello};
 use dosh::protocol::{
-    self, AttachOk, CLIENT_TO_SERVER, Frame, Header, Input, PacketKind, ResumeRequest,
-    SERVER_TO_CLIENT,
+    self, AttachOk, AttachReject, CLIENT_TO_SERVER, Frame, Header, Input, NativeAuthOkBody,
+    NativeClientHelloBody, NativeServerHelloBody, NativeUserAuthBody, PacketKind, ResumeRequest,
+    SERVER_TO_CLIENT, StreamData, StreamOpen, StreamOpenOk,
 };
+use ed25519_dalek::{SigningKey, VerifyingKey};
 use rand::rngs::StdRng;
 use rand::{Rng, SeedableRng};
 
@@ -79,6 +85,23 @@ persist_sessions = false
     config
 }
 
+fn authorize_ed25519_key(dir: &tempfile::TempDir, signing_key: &SigningKey) {
+    let verifying = VerifyingKey::from(signing_key);
+    let mut blob = Vec::new();
+    write_ssh_string(&mut blob, b"ssh-ed25519");
+    write_ssh_string(&mut blob, verifying.as_bytes());
+    fs::write(
+        dir.path().join("authorized_keys"),
+        format!("ssh-ed25519 {} hostile-test\n", STANDARD.encode(blob)),
+    )
+    .unwrap();
+}
+
+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 start_server(dir: &tempfile::TempDir, config: &std::path::Path) -> Child {
     let server = env!("CARGO_BIN_EXE_dosh-server");
     let child = Command::new(server)
@@ -402,6 +425,122 @@ fn attach_through_relay(
     }
 }
 
+struct NativeAttached {
+    socket: UdpSocket,
+    ok: NativeAuthOk,
+}
+
+fn native_attach_through_relay(
+    relay: &Relay,
+    signing_key: &SigningKey,
+    requested_forwardings: Vec<ForwardingRequest>,
+) -> NativeAttached {
+    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+    socket
+        .set_read_timeout(Some(Duration::from_millis(500)))
+        .unwrap();
+    let (client_secret, client_public) = native::generate_native_ephemeral();
+    let hello = NativeClientHello {
+        protocol_version: native::NATIVE_PROTOCOL_VERSION,
+        client_random: crypto::random_32(),
+        client_ephemeral_public: client_public,
+        requested_host: "127.0.0.1".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: native::supported_user_key_algorithms(),
+        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: hello.clone(),
+        })
+        .unwrap(),
+    )
+    .unwrap();
+    let hello_deadline = Instant::now() + Duration::from_secs(5);
+    let server_hello = loop {
+        socket.send_to(&packet, relay.front_addr()).unwrap();
+        let mut buf = [0u8; 65535];
+        match socket.recv_from(&mut buf) {
+            Ok((n, _)) => {
+                let packet = protocol::decode(&buf[..n]).unwrap();
+                match packet.header.kind {
+                    PacketKind::NativeServerHello => {
+                        let body: NativeServerHelloBody =
+                            protocol::from_body(&packet.body).unwrap();
+                        native::verify_server_hello(&hello, &body.hello).unwrap();
+                        break body.hello;
+                    }
+                    PacketKind::AttachReject => {
+                        let reject: AttachReject = protocol::from_body(&packet.body).unwrap();
+                        panic!("native hello rejected: {}", reject.reason);
+                    }
+                    kind => panic!("unexpected native hello response: {kind:?}"),
+                }
+            }
+            Err(_) if Instant::now() < hello_deadline => {}
+            Err(err) => panic!("native server hello timed out: {err}"),
+        }
+    };
+
+    let session_key = native::derive_native_session_key(
+        &client_secret,
+        server_hello.server_ephemeral_public,
+        &hello,
+        &server_hello,
+    )
+    .unwrap();
+    let auth =
+        native::sign_user_auth(signing_key, &hello, &server_hello, requested_forwardings).unwrap();
+    let mut pending_id = [0u8; 16];
+    pending_id.copy_from_slice(&server_hello.auth_challenge[..16]);
+    let auth_packet = protocol::encode_encrypted(
+        PacketKind::NativeUserAuth,
+        pending_id,
+        2,
+        1,
+        &session_key,
+        CLIENT_TO_SERVER,
+        &protocol::to_body(&NativeUserAuthBody { auth }).unwrap(),
+    )
+    .unwrap();
+    let auth_deadline = Instant::now() + Duration::from_secs(5);
+    let ok = loop {
+        socket.send_to(&auth_packet, relay.front_addr()).unwrap();
+        let mut buf = [0u8; 65535];
+        match socket.recv_from(&mut buf) {
+            Ok((n, _)) => {
+                let packet = protocol::decode(&buf[..n]).unwrap();
+                match packet.header.kind {
+                    PacketKind::NativeAuthOk => {
+                        let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT)
+                            .unwrap();
+                        let body: NativeAuthOkBody = protocol::from_body(&plain).unwrap();
+                        break body.ok;
+                    }
+                    PacketKind::AttachReject => {
+                        let reject: AttachReject = protocol::from_body(&packet.body).unwrap();
+                        panic!("native auth rejected: {}", reject.reason);
+                    }
+                    kind => panic!("unexpected native auth response: {kind:?}"),
+                }
+            }
+            Err(_) if Instant::now() < auth_deadline => {}
+            Err(err) => panic!("native auth ok timed out: {err}"),
+        }
+    };
+    NativeAttached { socket, ok }
+}
+
 fn send_input(
     socket: &UdpSocket,
     relay: &Relay,
@@ -431,6 +570,105 @@ fn send_raw(socket: &UdpSocket, relay: &Relay, packet: &[u8]) {
     socket.send_to(packet, relay.front_addr()).unwrap();
 }
 
+fn send_stream_packet(
+    socket: &UdpSocket,
+    relay: &Relay,
+    ok: &NativeAuthOk,
+    kind: PacketKind,
+    seq: u64,
+    ack: u64,
+    body: Vec<u8>,
+) {
+    let packet = protocol::encode_encrypted(
+        kind,
+        ok.client_id,
+        seq,
+        ack,
+        &ok.session_key,
+        CLIENT_TO_SERVER,
+        &body,
+    )
+    .unwrap();
+    socket.send_to(&packet, relay.front_addr()).unwrap();
+}
+
+fn recv_stream_packet<T: serde::de::DeserializeOwned>(
+    socket: &UdpSocket,
+    key: &[u8; 32],
+    kind: PacketKind,
+) -> Option<(Header, T)> {
+    let mut buf = [0u8; 65535];
+    let (n, _) = socket.recv_from(&mut buf).ok()?;
+    let packet = protocol::decode(&buf[..n]).ok()?;
+    if packet.header.kind != kind {
+        return None;
+    }
+    let plain = protocol::decrypt_body(&packet, key, SERVER_TO_CLIENT).ok()?;
+    let body = protocol::from_body(&plain).ok()?;
+    Some((packet.header, body))
+}
+
+fn wait_for_stream_open_ok(
+    socket: &UdpSocket,
+    key: &[u8; 32],
+    stream_id: u64,
+    millis: u64,
+) -> Option<Header> {
+    let deadline = Instant::now() + Duration::from_millis(millis);
+    while Instant::now() < deadline {
+        if let Some((header, ok)) =
+            recv_stream_packet::<StreamOpenOk>(socket, key, PacketKind::StreamOpenOk)
+        {
+            if ok.stream_id == stream_id {
+                return Some(header);
+            }
+        }
+    }
+    None
+}
+
+fn start_tcp_collector() -> (u16, Receiver<Vec<u8>>) {
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let (tx, rx) = channel();
+    thread::spawn(move || {
+        let Ok((mut stream, _)) = listener.accept() else {
+            return;
+        };
+        stream
+            .set_read_timeout(Some(Duration::from_millis(100)))
+            .unwrap();
+        let deadline = Instant::now() + Duration::from_secs(5);
+        let mut buf = [0u8; 1024];
+        while Instant::now() < deadline {
+            match stream.read(&mut buf) {
+                Ok(0) => return,
+                Ok(n) => {
+                    let _ = tx.send(buf[..n].to_vec());
+                }
+                Err(ref err)
+                    if err.kind() == std::io::ErrorKind::WouldBlock
+                        || err.kind() == std::io::ErrorKind::TimedOut => {}
+                Err(_) => return,
+            }
+        }
+    });
+    (port, rx)
+}
+
+fn collect_tcp(rx: &Receiver<Vec<u8>>, millis: u64) -> Vec<u8> {
+    let deadline = Instant::now() + Duration::from_millis(millis);
+    let mut out = Vec::new();
+    while Instant::now() < deadline {
+        match rx.recv_timeout(Duration::from_millis(50)) {
+            Ok(chunk) => out.extend_from_slice(&chunk),
+            Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
+            Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
+        }
+    }
+    out
+}
+
 fn recv_frame(socket: &UdpSocket, key: &[u8; 32]) -> Option<(Header, Frame)> {
     let mut buf = [0u8; 65535];
     let (n, _) = socket.recv_from(&mut buf).ok()?;
@@ -603,6 +841,124 @@ fn duplicated_and_replayed_input_is_applied_at_most_once() {
     );
 }
 
+#[test]
+fn forwarded_stream_data_survives_reorder_and_rejects_replay() {
+    let dir = tempfile::tempdir().unwrap();
+    let port = free_udp_port();
+    let config = write_server_config(&dir, port);
+    let signing_key = SigningKey::from_bytes(&[42u8; 32]);
+    authorize_ed25519_key(&dir, &signing_key);
+    let (target_port, tcp_rx) = start_tcp_collector();
+    let mut server = start_server(&dir, &config);
+    let relay = Relay::spawn(port, 0xF0E0D0u64);
+
+    let attached = native_attach_through_relay(
+        &relay,
+        &signing_key,
+        vec![ForwardingRequest {
+            kind: ForwardingKind::Local,
+            bind_host: Some("127.0.0.1".to_string()),
+            listen_port: 0,
+            target_host: Some("127.0.0.1".to_string()),
+            target_port: Some(target_port),
+        }],
+    );
+    let stream_id = 44;
+    let open = StreamOpen {
+        stream_id,
+        target_host: "127.0.0.1".to_string(),
+        target_port,
+    };
+    send_stream_packet(
+        &attached.socket,
+        &relay,
+        &attached.ok,
+        PacketKind::StreamOpen,
+        3,
+        attached.ok.initial_seq,
+        protocol::to_body(&open).unwrap(),
+    );
+    let open_ok =
+        wait_for_stream_open_ok(&attached.socket, &attached.ok.session_key, stream_id, 3000)
+            .expect("stream did not open");
+
+    // Swap the first two StreamData packets in flight. The server's replay
+    // window must accept both out-of-order packets and write both to the TCP
+    // target exactly once.
+    relay.arm_reorder();
+    send_stream_packet(
+        &attached.socket,
+        &relay,
+        &attached.ok,
+        PacketKind::StreamData,
+        4,
+        open_ok.seq,
+        protocol::to_body(&StreamData {
+            stream_id,
+            bytes: b"A".to_vec(),
+        })
+        .unwrap(),
+    );
+    relay.arm_reorder();
+    send_stream_packet(
+        &attached.socket,
+        &relay,
+        &attached.ok,
+        PacketKind::StreamData,
+        5,
+        open_ok.seq,
+        protocol::to_body(&StreamData {
+            stream_id,
+            bytes: b"B".to_vec(),
+        })
+        .unwrap(),
+    );
+
+    let mut seen = collect_tcp(&tcp_rx, 1500);
+    assert!(
+        seen.contains(&b'A') && seen.contains(&b'B'),
+        "reordered stream bytes did not both reach TCP target: {:?}",
+        String::from_utf8_lossy(&seen)
+    );
+
+    // Now send one encrypted StreamData packet repeatedly, with relay-level
+    // duplication enabled too. This is a true replay: same sequence, same nonce,
+    // same ciphertext. It must be applied to the TCP target at most once.
+    let replayed = protocol::encode_encrypted(
+        PacketKind::StreamData,
+        attached.ok.client_id,
+        6,
+        open_ok.seq,
+        &attached.ok.session_key,
+        CLIENT_TO_SERVER,
+        &protocol::to_body(&StreamData {
+            stream_id,
+            bytes: b"X".to_vec(),
+        })
+        .unwrap(),
+    )
+    .unwrap();
+    relay.set_dup(100);
+    for _ in 0..12 {
+        send_raw(&attached.socket, &relay, &replayed);
+        thread::sleep(Duration::from_millis(25));
+    }
+    relay.set_dup(0);
+
+    seen.extend(collect_tcp(&tcp_rx, 1500));
+    let replay_count = seen.iter().filter(|byte| **byte == b'X').count();
+
+    drop(relay);
+    let _ = server.kill();
+    let _ = server.wait();
+
+    assert!(
+        replay_count <= 1,
+        "replayed/duplicated StreamData was applied {replay_count} times: {:?}",
+        String::from_utf8_lossy(&seen)
+    );
+}
+
 #[test]
 fn stale_packets_after_resume_are_ignored_not_fatal() {
     let dir = tempfile::tempdir().unwrap();

tests/integration_smoke.rs

@@ -1495,6 +1495,69 @@ fn resume_updates_udp_endpoint_for_roaming() {
     );
 }
 
+#[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::<u64>().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;