haematite-cli 0.21.12

use super::assistant_test_support::*;
use serde_json::{Value, json};
use std::{path::PathBuf, sync::Arc};

// These are the launch inputs consumed by the source-included transport.
// Keeping the transport private to its seat does not require another wire implementation.
mod config {
    use super::PathBuf;
    pub struct Seat {
        pub place: Place,
        pub store: String,
        pub harness: Harness,
    }
    pub enum Place {
        Local { door_socket: PathBuf },
        Network { door_url: url::Url },
    }
    pub struct Harness {
        pub acp_frame_bytes: usize,
        pub write_queue: usize,
        pub content_bytes: usize,
        pub session_timeout_ms: u64,
        pub id: String,
    }
}
mod tools {
    pub struct Identity {
        pub agent: String,
        pub pass: String,
    }
}
mod handshake {
    use super::Value;
    pub struct Session {
        pub resumable: bool,
        pub version: Version,
        pub info: Value,
    }
    #[derive(Clone, Copy)]
    pub enum Version {
        Two,
    }
    impl Version {
        pub const fn number(self) -> u16 {
            match self {
                Self::Two => 2,
            }
        }
    }
}
mod door_link {
    include!("../assistant/door_link.rs");
}

#[tokio::test(flavor = "current_thread")]
async fn real_seat_link_and_real_door_agree_hello_welcome_op_ack_item() -> Result {
    let f = Fixture::new(json!({}))?;
    let seat = config::Seat {
        place: config::Place::Local {
            door_socket: f.files.path().join("assistant.sock"),
        },
        store: "wards".into(),
        harness: harness(),
    };
    let identity = tools::Identity {
        agent: "agent-assistant".into(),
        pass: agent_token("agent-pass")?,
    };
    let session = handshake::Session {
        resumable: true,
        version: handshake::Version::Two,
        info: json!({"version":"fixture"}),
    };
    let mut link = door_link::Link::open(&seat, &identity, &session).await?;
    assert_eq!(link.person, PERSON);
    let mut events = Stream::open(f.answer("GET", "/events", &json!({})))?;
    events.event()?;
    let request = f.request("POST", "", &json!({"text":"real transport"}));
    let (door, site) = (Arc::clone(&f.door), Arc::clone(&f.site));
    let posted = std::thread::spawn(move || {
        written(super::route::answer(
            &door,
            &site,
            &request,
            crate::serve::pass::now(),
        ))
    });
    let op = link
        .input
        .recv()
        .await
        .ok_or_else(|| anyhow::anyhow!("no op"))??
        .ok_or_else(|| anyhow::anyhow!("closed"))?;
    assert_eq!(op["op"]["text"], "real transport");
    assert_eq!(op["op"]["person"], PERSON);
    link.output.send(json!({"ack":{"id":op["op"]["id"]}}))?;
    link.output.send(json!({"item":item(1)}))?;
    link.output.flush().await?;
    let response = posted.join().map_err(|_| anyhow::anyhow!("post panic"))??;
    assert!(response.starts_with("HTTP/1.1 202 "), "{response}");
    let item = events.event()?;
    assert!(
        item.contains("id: conversation:1") && item.contains("item 1"),
        "{item}"
    );
    events.close()?;
    Ok(())
}

fn harness() -> config::Harness {
    config::Harness {
        acp_frame_bytes: 262_144,
        write_queue: 16,
        content_bytes: 65_536,
        session_timeout_ms: 10_000,
        id: "codex".into(),
    }
}

/// HAEMNET-003 R2: the seat on another machine reaches the door at its
/// https origin, here a TLS front with the test authority's certificate
/// for `localhost` before the door's own loopback listener, as a site's
/// proxy stands before it. Its upgrade carries that origin, its pass alone
/// admits it, and an operation, its ack and its item cross as on the
/// socket. A seat that trusts only the system's roots is refused at TLS,
/// before any upgrade.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_seat_link_reaches_the_door_through_its_tls_front() -> Result {
    let fixtures =
        std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src/serve/network_fixtures");
    let front = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
    let origin = format!("https://localhost:{}", front.local_addr()?.port());
    let f = Fixture::new(json!({ "origin": origin }))?;
    let door_listener = std::net::TcpListener::bind("127.0.0.1:0")?;
    let door_address = door_listener.local_addr()?;
    let (door, site) = (Arc::clone(&f.door), Arc::clone(&f.site));
    std::thread::Builder::new()
        .name("door-tls-front-test".into())
        .spawn(move || {
            super::listen::serve(
                &door_listener,
                &door,
                &site,
                &super::listen::Limits {
                    max_connections: 8,
                    read_timeout: std::time::Duration::from_secs(10),
                    head_cap: 16_384,
                    body_cap: 65_536,
                },
            );
        })?;
    tls_front(front, door_address, &fixtures)?;
    let seat = config::Seat {
        place: config::Place::Network {
            door_url: url::Url::parse(&origin)?,
        },
        store: "wards".into(),
        harness: harness(),
    };
    let identity = tools::Identity {
        agent: "agent-assistant".into(),
        pass: agent_token("agent-pass")?,
    };
    let session = handshake::Session {
        resumable: true,
        version: handshake::Version::Two,
        info: json!({"version":"fixture"}),
    };

    let untrusted = crate::client::network::Trust::mozilla().map_err(anyhow::Error::msg)?;
    let refused = door_link::Link::connect_with(&seat, &identity, &session, Some(&untrusted))
        .await
        .err()
        .ok_or_else(|| anyhow::anyhow!("a door certificate no root vouches for was trusted"))?;
    assert!(
        refused.to_string().starts_with("assistant_door_unreached:"),
        "{refused}"
    );

    let trust = crate::client::network::Trust::only(&fixtures.join("ca.pem"))
        .map_err(anyhow::Error::msg)?;
    let mut link = door_link::Link::connect_with(&seat, &identity, &session, Some(&trust)).await?;
    link.welcome(&seat, &identity).await?;
    assert_eq!(link.person, PERSON);
    let request = f.request("POST", "", &json!({"text":"through the front"}));
    let (door, site) = (Arc::clone(&f.door), Arc::clone(&f.site));
    let posted = std::thread::spawn(move || {
        written(super::route::answer(
            &door,
            &site,
            &request,
            crate::serve::pass::now(),
        ))
    });
    let op = link
        .input
        .recv()
        .await
        .ok_or_else(|| anyhow::anyhow!("no op"))??
        .ok_or_else(|| anyhow::anyhow!("closed"))?;
    assert_eq!(op["op"]["text"], "through the front");
    assert_eq!(op["op"]["person"], PERSON);
    link.output.send(json!({"ack":{"id":op["op"]["id"]}}))?;
    link.output.send(json!({"item":item(1)}))?;
    link.output.flush().await?;
    let response = tokio::task::spawn_blocking(move || posted.join())
        .await?
        .map_err(|_| anyhow::anyhow!("post panic"))??;
    assert!(response.starts_with("HTTP/1.1 202 "), "{response}");
    Ok(())
}

/// The fixture authority's TLS for `localhost` on `front`, each connection
/// copied both ways to the door's loopback listener at `door_address`.
fn tls_front(
    front: tokio::net::TcpListener,
    door_address: std::net::SocketAddr,
    fixtures: &std::path::Path,
) -> Result {
    let acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(
        crate::serve::transport::network::server_config(
            &crate::serve::transport::network::TlsFiles {
                certificate_file: fixtures.join("server.pem"),
                key_file: fixtures.join("server.key"),
            },
        )
        .map_err(anyhow::Error::msg)?,
    ));
    tokio::spawn(async move {
        while let Ok((tcp, _)) = front.accept().await {
            let acceptor = acceptor.clone();
            tokio::spawn(async move {
                let Ok(mut tls) = acceptor.accept(tcp).await else {
                    return;
                };
                if let Ok(mut door) = tokio::net::TcpStream::connect(door_address).await
                    && let Err(error) = tokio::io::copy_bidirectional(&mut tls, &mut door).await
                {
                    eprintln!("tls front: {error}");
                }
            });
        }
    });
    Ok(())
}