haematite-cli 0.21.12

//! HAEMNET-003 R2: the door's `GET /assistant` WebSocket, served by the
//! door's own listener on loopback, with a seat that speaks the codec by
//! hand. The seat's own link over a TLS front is
//! `assistant_link_tests::the_seat_link_reaches_the_door_through_its_tls_front`.

use super::super::assistant_test_support::{Fixture, ORIGIN, PERSON, agent_token, item, written};
use super::super::listen::{Limits, serve};
use super::super::route::answer;
use super::{OPCODE_BINARY, OPCODE_TEXT, accept, read_frame, write_frame};
use serde_json::{Value, json};
use std::io::{BufRead as _, BufReader, Read as _, Write as _};
use std::net::{TcpListener, TcpStream};
use std::sync::Arc;
use std::time::Duration;

type Result<T = ()> = anyhow::Result<T>;

const KEY: &str = "dGhlIHNhbXBsZSBub25jZQ==";

/// The RFC 6455 section 1.3 example.
#[test]
fn the_accept_value_is_the_rfc_example() {
    assert_eq!(accept(KEY), "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=");
}

/// The fixture's door, served on a loopback port of its own.
fn door(bounds: Value) -> Result<(Fixture, std::net::SocketAddr)> {
    let f = Fixture::new(bounds)?;
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let address = listener.local_addr()?;
    let (door, site) = (Arc::clone(&f.door), Arc::clone(&f.site));
    std::thread::Builder::new()
        .name("door-network-test".into())
        .spawn(move || {
            serve(
                &listener,
                &door,
                &site,
                &Limits {
                    max_connections: 8,
                    read_timeout: Duration::from_secs(10),
                    head_cap: 16_384,
                    body_cap: 65_536,
                },
            );
        })?;
    Ok((f, address))
}

/// An upgrade of `/assistant` from `origin`; the answer's head and the
/// connection, positioned after it.
fn upgrade(address: std::net::SocketAddr, origin: &str) -> Result<(String, BufReader<TcpStream>)> {
    let mut stream = TcpStream::connect(address)?;
    stream.set_read_timeout(Some(Duration::from_secs(10)))?;
    write!(
        stream,
        "GET /assistant HTTP/1.1\r\nHost: localhost\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: {KEY}\r\nSec-WebSocket-Version: 13\r\nOrigin: {origin}\r\n\r\n"
    )?;
    let mut reader = BufReader::new(stream);
    let mut head = String::new();
    while !head.ends_with("\r\n\r\n") {
        anyhow::ensure!(
            reader.read_line(&mut head)? > 0,
            "closed in the head: {head}"
        );
    }
    Ok((head, reader))
}

fn send(reader: &mut BufReader<TcpStream>, opcode: u8, value: &Value) -> Result {
    write_frame(
        reader.get_mut(),
        opcode,
        &serde_json::to_vec(value)?,
        Some([1, 2, 3, 4]),
    )?;
    Ok(())
}

/// The door's next message as JSON; `None` at its close.
fn message(reader: &mut BufReader<TcpStream>) -> Result<Option<Value>> {
    let Some(frame) =
        read_frame(reader, 1 << 20, false).map_err(|failure| anyhow::anyhow!(failure.code))?
    else {
        return Ok(None);
    };
    if frame.opcode != OPCODE_BINARY {
        return Ok(None);
    }
    Ok(Some(serde_json::from_slice(&frame.payload)?))
}

fn hello(pass: &str) -> Result<Value> {
    let agent = if pass == "squatter" {
        "agent-squatter"
    } else {
        "agent-assistant"
    };
    let pass = if pass.is_empty() {
        String::new()
    } else {
        agent_token(pass)?
    };
    Ok(
        json!({"hello":{"agent":agent,"pass":pass,"store":"wards","resumable":true,"harness":{"id":"codex","version":"1","protocol":2}}}),
    )
}

/// A seat on the network is admitted by its pass alone, is sent the
/// operation a person posts, acknowledges it, and its item reaches the
/// person's stream: the socket's frames, one per binary message.
#[test]
fn a_seat_upgraded_on_the_network_is_admitted_by_its_pass_and_serves_a_turn() -> Result {
    let (f, address) = door(json!({}))?;
    let (head, mut seat) = upgrade(address, ORIGIN)?;
    assert!(head.starts_with("HTTP/1.1 101 "), "{head}");
    assert!(
        head.contains(&format!("Sec-WebSocket-Accept: {}", accept(KEY))),
        "{head}"
    );
    send(&mut seat, OPCODE_BINARY, &hello("agent-pass")?)?;
    assert_eq!(
        message(&mut seat)?,
        Some(json!({"welcome":{"person":PERSON}}))
    );

    let request = f.request("POST", "", &json!({"text":"over the network"}));
    let (door, site) = (Arc::clone(&f.door), Arc::clone(&f.site));
    let posted = std::thread::spawn(move || {
        written(answer(&door, &site, &request, crate::serve::pass::now()))
    });
    let op = message(&mut seat)?.ok_or_else(|| anyhow::anyhow!("no op"))?;
    assert_eq!(op["op"]["text"], "over the network");
    send(
        &mut seat,
        OPCODE_BINARY,
        &json!({"ack":{"id":op["op"]["id"]}}),
    )?;
    send(&mut seat, OPCODE_BINARY, &json!({"item":item(1)}))?;
    let response = posted
        .join()
        .map_err(|_| anyhow::anyhow!("post panicked"))??;
    assert!(response.starts_with("HTTP/1.1 202 "), "{response}");
    Ok(())
}

/// Each refused hello (another audience, another agent, no responsible
/// person, a grant naming another store, an expired pass, no pass) is
/// answered by the pass check's own name, and the seat is sent no
/// operation: the connection closes after the refusal.
#[test]
fn a_seat_with_a_wrong_pass_is_refused_by_name_and_sent_no_frame() -> Result {
    let (_f, address) = door(json!({}))?;
    for (pass, code) in [
        ("wrong-audience", "wrong_audience"),
        ("wrong-agent", "assistant_agent_mismatch"),
        ("unowned-agent", "contract_refused"),
        ("squatter", "seat_not_yours"),
        ("expired", "pass_expired"),
        ("", "contract_refused"),
    ] {
        let (head, mut seat) = upgrade(address, ORIGIN)?;
        assert!(head.starts_with("HTTP/1.1 101 "), "{head}");
        send(&mut seat, OPCODE_BINARY, &hello(pass)?)?;
        let refused =
            message(&mut seat)?.ok_or_else(|| anyhow::anyhow!("{pass}: closed unanswered"))?;
        assert_eq!(refused["refused"]["code"], code, "{pass}: {refused}");
        assert_eq!(
            message(&mut seat)?,
            None,
            "{pass}: nothing after the refusal"
        );
    }
    Ok(())
}

/// An upgrade from another origin is refused before any WebSocket exists.
#[test]
fn an_upgrade_from_a_foreign_origin_is_refused_by_name() -> Result {
    let (_f, address) = door(json!({}))?;
    let (head, mut answer) = upgrade(address, "https://elsewhere.example")?;
    assert!(head.starts_with("HTTP/1.1 403 "), "{head}");
    let mut body = String::new();
    answer.read_to_string(&mut body)?;
    assert!(body.contains("origin_refused"), "{body}");
    Ok(())
}

/// A text message, and a message over `frame_bytes`, are refused by name,
/// and the door closes.
#[test]
fn a_text_message_and_an_oversized_one_are_refused_by_name() -> Result {
    let (_f, address) = door(json!({"frame_bytes": 4096}))?;
    let (_, mut seat) = upgrade(address, ORIGIN)?;
    send(&mut seat, OPCODE_TEXT, &hello("agent-pass")?)?;
    let refused = message(&mut seat)?.ok_or_else(|| anyhow::anyhow!("text: closed unanswered"))?;
    assert_eq!(
        refused["refused"]["code"], "assistant_frame_invalid",
        "{refused}"
    );
    assert_eq!(message(&mut seat)?, None);

    let (_, mut seat) = upgrade(address, ORIGIN)?;
    write_frame(
        seat.get_mut(),
        OPCODE_BINARY,
        &vec![b'x'; 5000],
        Some([9, 9, 9, 9]),
    )?;
    let refused = message(&mut seat)?.ok_or_else(|| anyhow::anyhow!("bound: closed unanswered"))?;
    assert_eq!(refused["refused"]["code"], "assistant_bound", "{refused}");
    assert!(
        refused["refused"]["message"]
            .as_str()
            .is_some_and(|words| words.starts_with("frame_bytes=4096")),
        "{refused}"
    );
    assert_eq!(message(&mut seat)?, None);
    Ok(())
}