nordisk/Skidbladnir

crates/installer/plugin-door-wire/tests/slow_door.rs

//! **R1 (gunnar release-3 estate-box check 1, 3 runs in 6): the POST was never all sent.**
//!
//! MEASURED with the door's stages at DEBUG: the primary's `POST /register/gunnar` (an
//! 850 KB module, base64) was accepted and its TLS handshake done on vm1 — and the handler
//! never ran; 180 s later the client gave up with `no HTTP answer`. Both ends ESTABLISHED,
//! both kernel queues EMPTY: the last of the request sat in the CLIENT's TLS buffer.
//! `tokio_rustls`'s `poll_write` answers `Ready(n)` once rustls TOOK the bytes, even when
//! the socket would block before they were all sealed out — and `exchange` went straight
//! from `write_all` to `read_to_end` without a `flush`, so the rest was never written: the
//! door waited for the body, the client for the answer. On loopback with a door that reads
//! at once the socket never blocks; a door that is slow to read (vm1 busy at the close)
//! fills the socket, and the request stops short.
//!
//! The door here reads NOTHING for a moment, through a small receive buffer, then reads the
//! whole request and answers: the client must get its answer.

use std::net::SocketAddr;
use std::time::Duration;

use nordisk_pairing::mtls::{self, InstallerCa, Issued};
use plugin_door_wire::{Registered, SelfRegister};
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};

const DAY: Duration = Duration::from_secs(24 * 3600);

fn boxed(ca: &InstallerCa, name: &str) -> Issued {
    let (csr, key) = mtls::box_request().expect("csr");
    Issued { cert_pem: ca.sign_box(&csr, name, DAY).expect("sign"), key_pem: key }
}

/// A door that waits `pause` before it reads, with a small receive buffer, then reads the
/// head and Content-Length bytes and answers 200 with a Registered.
async fn slow_door(ca: &InstallerCa, door_cert: Issued, pause: Duration) -> SocketAddr {
    let cfg = mtls::server_config(&ca.cert_pem, &door_cert).expect("server config");
    let socket = tokio::net::TcpSocket::new_v4().unwrap();
    socket.set_recv_buffer_size(4096).unwrap();
    socket.bind("127.0.0.1:0".parse().unwrap()).unwrap();
    let listener = socket.listen(8).unwrap();
    let at = listener.local_addr().unwrap();
    tokio::spawn(async move {
        loop {
            let Ok((tcp, _)) = listener.accept().await else { return };
            let acceptor = tokio_rustls::TlsAcceptor::from(cfg.clone());
            tokio::spawn(async move {
                let Ok(mut tls) = acceptor.accept(tcp).await else { return };
                tokio::time::sleep(pause).await;
                let mut buf = Vec::new();
                let mut chunk = [0u8; 65536];
                loop {
                    let n = tls.read(&mut chunk).await.unwrap_or(0);
                    if n == 0 {
                        return;
                    }
                    buf.extend_from_slice(&chunk[..n]);
                    if let Some(i) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
                        let head = String::from_utf8_lossy(&buf[..i]).to_string();
                        let len: usize = head.lines().find_map(|l| l.strip_prefix("Content-Length: ")).and_then(|v| v.trim().parse().ok()).unwrap_or(0);
                        while buf.len() < i + 4 + len {
                            let n = tls.read(&mut chunk).await.unwrap_or(0);
                            if n == 0 {
                                return;
                            }
                            buf.extend_from_slice(&chunk[..n]);
                        }
                        break;
                    }
                }
                let json = serde_json::to_string(&Registered { outcome: "ADDED".into(), line: "guest ADDED [products.gunnar]".into(), monetize_pubkey: None }).unwrap();
                let out = format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{json}", json.len());
                let _ = tls.write_all(out.as_bytes()).await;
                let _ = tls.shutdown().await;
            });
        }
    });
    at
}

#[tokio::test(flavor = "current_thread")]
async fn a_module_sized_post_to_a_door_slow_to_read_is_all_sent_and_answered() {
    let ca = InstallerCa::mint("gunnar.test", DAY).unwrap();
    let door = slow_door(&ca, ca.issue_bootstrap(DAY).unwrap(), Duration::from_millis(500)).await;
    let main = boxed(&ca, "main");
    // Larger than any socket buffer between the two: the socket MUST block mid-body.
    let module = vec![0x42u8; 12 << 20];
    let req = SelfRegister { module: plugin_door_wire::b64(&module), signature: plugin_door_wire::b64(b"sig"), ticket_key: Some("ab".repeat(32)) };
    for i in 0..2 {
        let r = tokio::time::timeout(Duration::from_secs(20), plugin_door_wire::register_remote(door, &ca.cert_pem, &main, "gunnar", &req)).await;
        match r {
            Ok(Ok((200, a))) => assert!(a.taken(), "{a:?}"),
            other => panic!("exchange {i}: the POST was not answered within 20 s — the request never reached the door whole: {other:?}"),
        }
    }
}