liminal-server 0.25.4

//! The pre-admission inbound bound on the byte-stream transport: an
//! unadmitted peer's header declaring more than the bound is refused by name
//! before its body is buffered, and the negotiated bound applies once the
//! `Connect` has succeeded.

use std::error::Error;
use std::sync::{Arc, Mutex};

use liminal::protocol::{Frame, decode};

use super::super::pre_admission::{PRE_ADMISSION_FRAME_BOUND, PreAdmissionOversize};
use super::{ConnectionProcess, ProcessStatus, process_buffer};
use crate::server::connection::supervisor::runtime::ConnectionRuntime;
use crate::server::connection::worker_front_door::WorkerFrontDoorServices;

/// A generic `Publish` header (frame type 0x09) declaring a complete frame of
/// `complete_bytes`, with none of the body present.
fn header_declaring(complete_bytes: usize) -> Result<Vec<u8>, Box<dyn Error>> {
    let payload = u32::try_from(complete_bytes - 10)?;
    let mut header = vec![0x09, 0x00, 0x00, 0x00, 0x00, 0x01];
    header.extend_from_slice(&payload.to_be_bytes());
    Ok(header)
}

fn process() -> Result<(ConnectionProcess, std::net::TcpStream), Box<dyn Error>> {
    let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
    let client = std::net::TcpStream::connect(listener.local_addr()?)?;
    let (server, _) = listener.accept()?;
    let runtime = Arc::new(ConnectionRuntime::for_tests(Arc::new(
        WorkerFrontDoorServices::new(),
    )));
    let holder = Arc::new(Mutex::new(Some(server)));
    Ok((
        ConnectionProcess::from_holder(runtime, None, &holder, None),
        client,
    ))
}

fn run(process: &mut ConnectionProcess) -> Result<ProcessStatus, Box<dyn Error>> {
    let runtime = Arc::clone(&process.runtime);
    Ok(process_buffer(
        1,
        &runtime,
        &mut process.state,
        &mut process.buffer,
        &mut process.outbound,
    )?)
}

#[test]
fn unadmitted_header_over_the_bound_is_refused_by_name_before_its_body_is_held()
-> Result<(), Box<dyn Error>> {
    let (mut process, _client) = process()?;
    process.buffer = header_declaring(PRE_ADMISSION_FRAME_BOUND + 1)?;

    let status = run(&mut process)?;

    assert_eq!(status, ProcessStatus::Close);
    assert_eq!(
        process.buffer.len(),
        0,
        "no byte of the refused frame is held"
    );
    let queued = process.outbound.take_bytes();
    let (frame, consumed) = decode(&queued)?;
    assert_eq!(
        consumed,
        queued.len(),
        "exactly one refusal frame is queued"
    );
    let expected = PreAdmissionOversize {
        attempted: PRE_ADMISSION_FRAME_BOUND + 1,
        bound: PRE_ADMISSION_FRAME_BOUND,
    }
    .to_string();
    assert!(
        matches!(&frame, Frame::ConnectError { message: Some(message), .. } if *message == expected),
        "{frame:?}"
    );
    Ok(())
}

#[test]
fn unadmitted_header_at_the_bound_waits_for_its_body() -> Result<(), Box<dyn Error>> {
    let (mut process, _client) = process()?;
    process.buffer = header_declaring(PRE_ADMISSION_FRAME_BOUND)?;

    let status = run(&mut process)?;

    assert_eq!(status, ProcessStatus::Continue);
    assert_eq!(process.buffer.len(), 10, "only the header is held");
    Ok(())
}

#[test]
fn admitted_header_over_the_pre_admission_bound_waits_for_its_body() -> Result<(), Box<dyn Error>> {
    let (mut process, _client) = process()?;
    process.state.authenticated = true;
    process.buffer = header_declaring(PRE_ADMISSION_FRAME_BOUND + 1)?;

    let status = run(&mut process)?;

    assert_eq!(status, ProcessStatus::Continue);
    assert_eq!(process.buffer.len(), 10, "only the header is held");
    Ok(())
}