e2e.rs

//! End-to-end integration tests.
//!
//! These spin up the server on a tempdir UDS and drive it from a
//! handcrafted client that uses `recvmsg`/`sendmsg` directly so we can
//! verify cmsg fd-passing semantics.

use std::io::{IoSlice, IoSliceMut};
use std::mem::MaybeUninit;
use std::os::fd::{AsRawFd, BorrowedFd, OwnedFd};
use std::os::unix::net::UnixStream as StdUnixStream;

use http::StatusCode;
use hyper_uds::{Builder, Request, Response, service_fn};
use rustix::net::{
    RecvAncillaryBuffer, RecvAncillaryMessage, RecvFlags, SendAncillaryBuffer,
    SendAncillaryMessage, SendFlags,
};

/// Client helper: send a request (head + body + optional fds) over a
/// blocking std UnixStream using rustix sendmsg, then read until we
/// have a complete HTTP/1.1 response.
fn round_trip(
    stream: &StdUnixStream,
    request_bytes: &[u8],
    send_fds: &[BorrowedFd<'_>],
) -> (Vec<u8>, Vec<OwnedFd>) {
    // ---- send ----
    let bufs = [IoSlice::new(request_bytes)];
    if send_fds.is_empty() {
        let mut empty_anc = SendAncillaryBuffer::new(&mut []);
        rustix::net::sendmsg(stream, &bufs, &mut empty_anc, SendFlags::empty()).unwrap();
    } else {
        let mut space = [MaybeUninit::<u8>::uninit(); 256];
        let mut anc = SendAncillaryBuffer::new(&mut space);
        assert!(anc.push(SendAncillaryMessage::ScmRights(send_fds)));
        rustix::net::sendmsg(stream, &bufs, &mut anc, SendFlags::empty()).unwrap();
    }

    // ---- recv until headers and body fully arrive ----
    let mut all = Vec::new();
    let mut recv_fds: Vec<OwnedFd> = Vec::new();
    let mut content_length: Option<usize> = None;
    let mut header_end: Option<usize> = None;

    loop {
        let mut chunk = [0u8; 4096];
        let mut iov = [IoSliceMut::new(&mut chunk)];
        let mut space = [MaybeUninit::<u8>::uninit(); 256];
        let mut anc = RecvAncillaryBuffer::new(&mut space);
        let ret = rustix::net::recvmsg(stream, &mut iov, &mut anc, RecvFlags::empty()).unwrap();
        for msg in anc.drain() {
            if let RecvAncillaryMessage::ScmRights(fds) = msg {
                for fd in fds {
                    recv_fds.push(fd);
                }
            }
        }
        if ret.bytes == 0 {
            break;
        }
        all.extend_from_slice(&chunk[..ret.bytes]);

        if header_end.is_none()
            && let Some(pos) = find_double_crlf(&all)
        {
            header_end = Some(pos + 4);
            // parse content-length
            let head_str = std::str::from_utf8(&all[..pos]).unwrap();
            for line in head_str.split("\r\n") {
                if let Some(v) = line
                    .strip_prefix("content-length: ")
                    .or_else(|| line.strip_prefix("Content-Length: "))
                {
                    content_length = Some(v.trim().parse().unwrap());
                }
            }
        }

        if let (Some(he), Some(cl)) = (header_end, content_length)
            && all.len() >= he + cl
        {
            break;
        }
    }

    (all, recv_fds)
}

fn find_double_crlf(buf: &[u8]) -> Option<usize> {
    buf.windows(4).position(|w| w == b"\r\n\r\n")
}

#[test]
fn echo_plain_http() {
    let dir = tempfile::tempdir().unwrap();
    let sock = dir.path().join("s.sock");

    let listener = std::os::unix::net::UnixListener::bind(&sock).unwrap();

    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .enable_all()
        .build()
        .unwrap();

    // Client side: blocking std UnixStream
    let client = StdUnixStream::connect(&sock).unwrap();

    let req = b"POST /echo HTTP/1.1\r\nhost: localhost\r\ncontent-length: 11\r\n\r\nhello world";
    let handle = std::thread::spawn(move || round_trip(&client, req, &[]));

    rt.block_on(async move {
        listener.set_nonblocking(true).unwrap();
        let listener = tokio::net::UnixListener::from_std(listener).unwrap();
        let (stream, _) = listener.accept().await.unwrap();
        let std_stream = stream.into_std().unwrap();

        let svc = service_fn(|req: Request| async move {
            assert_eq!(req.method(), &http::Method::POST);
            assert_eq!(req.uri().path(), "/echo");
            let body = req.into_body();
            Ok::<_, std::convert::Infallible>(
                Response::builder()
                    .status(StatusCode::OK)
                    .body(body)
                    .build(),
            )
        });

        Builder::new()
            .keep_alive(false)
            .serve_connection(std_stream, svc)
            .await
            .unwrap();
    });
    let (resp_bytes, fds) = handle.join().unwrap();

    assert!(fds.is_empty());
    let resp = String::from_utf8(resp_bytes).unwrap();
    assert!(resp.starts_with("HTTP/1.1 200"), "got: {resp}");
    assert!(resp.contains("content-length: 11"));
    assert!(resp.ends_with("hello world"));
}

#[test]
fn fd_passing_round_trip() {
    let dir = tempfile::tempdir().unwrap();
    let sock = dir.path().join("s.sock");

    let listener = std::os::unix::net::UnixListener::bind(&sock).unwrap();

    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .enable_all()
        .build()
        .unwrap();

    // Client: prepare an fd pointing at a temp file.
    let payload_file = tempfile::NamedTempFile::new().unwrap();
    std::fs::write(payload_file.path(), b"hello-from-fd").unwrap();
    let payload = std::fs::File::open(payload_file.path()).unwrap();

    let client = StdUnixStream::connect(&sock).unwrap();
    let req = b"POST /open HTTP/1.1\r\nhost: localhost\r\ncontent-length: 0\r\nx-fd: input\r\n\r\n";

    let payload_fd_raw = payload.as_raw_fd();
    let handle = std::thread::spawn(move || {
        let borrowed = unsafe { BorrowedFd::borrow_raw(payload_fd_raw) };
        round_trip(&client, req, &[borrowed])
    });

    rt.block_on(async move {
        listener.set_nonblocking(true).unwrap();
        let listener = tokio::net::UnixListener::from_std(listener).unwrap();
        let (stream, _) = listener.accept().await.unwrap();
        let std_stream = stream.into_std().unwrap();

        let svc = service_fn(|mut req: Request| async move {
            // Verify we got the fd named "input"
            assert_eq!(req.fd_names(), &["input".to_string()]);
            assert_eq!(req.fds().len(), 1);

            // Read the contents of the fd to confirm it's the
            // file we expect.
            let mut fds = req.take_fds();
            let mut received = Vec::new();
            use std::io::Read;
            let owned: OwnedFd = fds.remove(0);
            let mut f = std::fs::File::from(owned);
            f.read_to_end(&mut received).unwrap();

            assert_eq!(received, b"hello-from-fd");

            // Echo back a *different* fd so we can verify direction
            // works both ways.
            let tmp = tempfile::NamedTempFile::new().unwrap();
            std::fs::write(tmp.path(), b"server-reply").unwrap();
            let f = std::fs::File::open(tmp.path()).unwrap();
            let resp_fd: OwnedFd = f.into();

            Ok::<_, std::convert::Infallible>(
                Response::builder()
                    .status(StatusCode::OK)
                    .fd("reply", resp_fd)
                    .body(bytes::Bytes::from_static(b"ok"))
                    .build(),
            )
        });

        Builder::new()
            .keep_alive(false)
            .serve_connection(std_stream, svc)
            .await
            .unwrap();
    });
    let (resp_bytes, fds) = handle.join().unwrap();
    drop(payload);

    assert_eq!(fds.len(), 1, "expected one returned fd");
    let resp = String::from_utf8(resp_bytes).unwrap();
    assert!(resp.contains("HTTP/1.1 200"), "got: {resp}");
    assert!(
        resp.to_ascii_lowercase().contains("x-fd: reply"),
        "got: {resp}"
    );

    // Read content of the returned fd to confirm it's the server's
    // reply file.
    use std::io::Read;
    let mut f = std::fs::File::from(fds.into_iter().next().unwrap());
    let mut s = Vec::new();
    f.read_to_end(&mut s).unwrap();
    assert_eq!(s, b"server-reply");
}