haematite-cli 0.21.1

The haematite command: serve a store, its door and Studio, answer MCP, query with jq, and host the store's assistant seat
Documentation
use std::io::Write;
use std::net::{Shutdown, TcpStream};
use std::num::NonZeroU64;
use std::sync::mpsc;
use std::time::Duration;

use super::fixtures::{LIST, Reply, Server, TestResult, config, refusal, request};
use super::listener;
use super::native_call_fixture::Store;

fn start(
    store: &Store,
) -> Result<(Server, mpsc::Receiver<String>), Box<dyn std::error::Error + Send + Sync>> {
    let (logged, lines) = mpsc::channel();
    let server = listener::Server::new(
        config()?,
        store.path.clone(),
        Duration::from_secs(1),
        131_072,
        1,
    )?
    .with_diagnostics(logged)?;
    Ok((Server::from_mcp(server)?, lines))
}

#[test]
fn an_unauthenticated_large_body_receives_the_whole_challenge_and_refusal() -> TestResult {
    let store = Store::start()?;
    let mut config = config()?;
    config.resource_metadata_url = Some("https://identity.example/resource-metadata".to_owned());
    let mcp = Server::mcp(config, store.path.clone(), 131_072, 1)?;
    let response = Reply::read(request(&mcp, "POST", &"x".repeat(65_536), false, &[])?)?;
    refusal(&response, 401, "mcp_sign_in_required")?;
    assert_eq!(
        response
            .headers
            .get("www-authenticate")
            .ok_or("challenge missing")?
            .to_str()?,
        "Bearer resource_metadata=\"https://identity.example/resource-metadata\""
    );
    assert!(store.asked()?.is_empty());
    mcp.finish()?;
    store.finish()
}

fn raw_refusal(bytes: &str, status: u16, code: &str, has_cause: bool) -> TestResult {
    let store = Store::start()?;
    let (mcp, lines) = start(&store)?;
    let mut socket = TcpStream::connect(mcp.address)?;
    socket.write_all(bytes.as_bytes())?;
    let response = Reply::read(socket)?;
    refusal(&response, status, code)?;
    let line = lines.try_recv()?;
    let body: serde_json::Value = serde_json::from_slice(&response.body)?;
    let message = body["message"].as_str().ok_or("reason missing")?;
    assert!(line.contains(code));
    assert_eq!(line.matches(message).count(), 1);
    assert_eq!(line.contains("; cause="), has_cause);
    assert_eq!(line.lines().count(), 1);
    assert!(store.asked()?.is_empty());
    mcp.finish()?;
    store.finish()
}

#[test]
fn headers_over_the_byte_bound_refuse_431_before_sign_in() -> TestResult {
    raw_refusal(
        &format!(
            concat!(
                "POST /mcp HTTP/1.1\r\nHost: fixture\r\nPadding: {}\r\n",
                "Content-Length: 0\r\n\r\n"
            ),
            "x".repeat(16_384)
        ),
        431,
        "mcp_http_headers_too_large",
        false,
    )
}

#[test]
fn expect_negotiation_refuses_417_before_sign_in() -> TestResult {
    raw_refusal(
        concat!(
            "POST /mcp HTTP/1.1\r\nHost: fixture\r\n",
            "Expect: 100-continue\r\nContent-Length: 0\r\n\r\n"
        ),
        417,
        "mcp_http_expect_unsupported",
        false,
    )
}

#[test]
fn an_unknown_route_refuses_404_before_sign_in() -> TestResult {
    raw_refusal(
        "POST /other HTTP/1.1\r\nHost: fixture\r\nContent-Length: 0\r\n\r\n",
        404,
        "mcp_http_route_missing",
        false,
    )
}

#[test]
fn an_incomplete_header_eof_is_named_and_logged_once() -> TestResult {
    let store = Store::start()?;
    let (mcp, lines) = start(&store)?;
    let mut socket = TcpStream::connect(mcp.address)?;
    socket.write_all(b"POST /mcp HTTP/1.1\r\nHost: fixture")?;
    socket.shutdown(Shutdown::Write)?;
    refusal(&Reply::read(socket)?, 400, "mcp_http_header_unreadable")?;
    let lines = lines.try_iter().collect::<Vec<_>>();
    assert_eq!(lines.len(), 1);
    assert!(lines[0].contains("mcp_http_header_unreadable"));
    assert!(store.asked()?.is_empty());
    mcp.finish()?;
    store.finish()
}

#[test]
fn a_partial_body_expires_on_the_configured_body_deadline() -> TestResult {
    let store = Store::start()?;
    let mut config = config()?;
    config.body_timeout_ms = NonZeroU64::new(50).ok_or("body timeout missing")?;
    let mcp = Server::mcp(config, store.path.clone(), 131_072, 1)?;
    let mut socket = TcpStream::connect(mcp.address)?;
    write!(
        socket,
        "POST /mcp HTTP/1.1\r\nHost: fixture\r\nContent-Length: {}\r\nContent-Type: application/json\r\nAccept: application/json, text/event-stream\r\nAuthorization: Bearer fixture-token\r\n\r\n{{",
        LIST.len()
    )?;
    refusal(&Reply::read(socket)?, 408, "mcp_http_body_timeout")?;
    assert!(store.asked()?.is_empty());
    mcp.finish()?;
    store.finish()
}

/// What the store behind the HTTP MCP does with a presented pass.
#[derive(Clone, Copy)]
enum Behind {
    Accepting,
    Refusing(&'static str),
    Absent,
}

type LogCase = (
    Behind,
    bool,
    Vec<(&'static str, &'static str)>,
    u16,
    &'static str,
);

fn logged_case((behind, token, extra, status, code): LogCase) -> TestResult {
    let refused = match behind {
        Behind::Refusing(name) => Some(name),
        Behind::Accepting | Behind::Absent => None,
    };
    let store = match behind {
        Behind::Accepting | Behind::Absent => Store::start()?,
        Behind::Refusing(code) => Store::refusing(code)?,
    };
    let (logged, lines) = mpsc::channel();
    let directory = tempfile::tempdir()?;
    let socket = match behind {
        Behind::Absent => directory.path().join("absent.sock"),
        Behind::Accepting | Behind::Refusing(_) => store.path.clone(),
    };
    let server = listener::Server::new(config()?, socket, Duration::from_secs(1), 131_072, 1)?
        .with_diagnostics(logged)?;
    let mcp = Server::from_mcp(server)?;
    let response = Reply::read(request(&mcp, "POST", LIST, token, &extra)?)?;
    refusal(&response, status, code)?;
    let line = lines.try_recv()?;
    let body: serde_json::Value = serde_json::from_slice(&response.body)?;
    let message = body["message"].as_str().ok_or("reason missing")?;
    assert!(line.contains(code));
    assert_eq!(line.matches(message).count(), 1);
    // Only a store's refusal adds a cause, and that cause is its plain name.
    match refused {
        Some(name) => assert!(line.ends_with(&format!("; cause=store:{name}")), "{line}"),
        None => assert!(!line.contains("; cause="), "{line}"),
    }
    assert!(!line.contains("fixture-token"));
    assert_eq!(line.lines().count(), 1);
    mcp.finish()?;
    store.finish()
}

#[test]
fn sign_in_protocol_and_store_refusals_have_safe_named_log_lines() -> TestResult {
    for case in [
        (
            Behind::Accepting,
            false,
            Vec::new(),
            401,
            "mcp_sign_in_required",
        ),
        (
            Behind::Accepting,
            true,
            vec![("mcp-protocol-version", "invalid")],
            400,
            "mcp_http_protocol_version_unsupported",
        ),
        (
            Behind::Refusing("pass_expired"),
            true,
            Vec::new(),
            401,
            "mcp_pass_refused",
        ),
        (
            Behind::Refusing("lys_could_not_be_asked"),
            true,
            Vec::new(),
            503,
            "mcp_store_unavailable",
        ),
        (
            Behind::Absent,
            true,
            Vec::new(),
            503,
            "mcp_store_unavailable",
        ),
    ] {
        logged_case(case)?;
    }
    Ok(())
}

#[test]
fn a_configured_path_is_the_only_one_answered() -> TestResult {
    let store = Store::start()?;
    let mut config = config()?;
    config.path = "/mcp/studio".to_owned();
    config.validate()?;
    let mcp = Server::mcp(config, store.path.clone(), 131_072, 1)?;
    for (path, status) in [("/mcp", 404), ("/mcp/studio/", 404), ("/mcp/studio", 200)] {
        let mut socket = TcpStream::connect(mcp.address)?;
        socket.write_all(
            format!(
                "POST {path} HTTP/1.1\r\nHost: fixture\r\nContent-Type: application/json\r\nAccept: application/json, text/event-stream\r\nAuthorization: Bearer fixture-token\r\nContent-Length: {}\r\n\r\n{LIST}",
                LIST.len()
            )
            .as_bytes(),
        )?;
        let response = Reply::read(socket)?;
        assert_eq!(response.status, status, "{path}");
        if status == 404 {
            refusal(&response, 404, "mcp_http_route_missing")?;
            assert!(store.asked()?.is_empty(), "{path} reached the store");
        }
    }
    assert!(
        !store.asked()?.is_empty(),
        "the configured path never reached the store"
    );
    mcp.finish()?;
    store.finish()
}