liminal-cli 0.3.0

//! SRV-029 R9 over a real TCP server: a ticket on a `[lys]` table calls a
//! worker with `pass_for`, and the worker's body carries `pass`, a pass the
//! (fake) Lys agent route minted for the asked audience with the calling
//! agent as its holder. The server never forwards the ticket's own liminal
//! pass; a call without `pass_for` carries no pass, and an input that already
//! carries one is refused before anything is sent.

use std::error::Error;
use std::io::Write as _;
use std::net::{SocketAddr, TcpListener};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use base64::Engine as _;
use serde_json::{Value, json};

use crate::lifecycle::read_request;
use crate::network::{RunningServer, lys_pass_aud};
use crate::{Host, SUBJECT, Ticket, answering_worker, test_persistence};

/// The agent the ticket asks Lys for passes as.
const AGENT: &str = "agent-under-test";
/// The audience the call names.
const AUDIENCE: &str = "haematite";
/// PKCS#8 v1 DER for an ed25519 private key, before its 32-byte seed.
const PKCS8_ED25519_PREFIX: [u8; 16] = [
    0x30, 0x2e, 0x02, 0x01, 0x00, 0x30, 0x05, 0x06, 0x03, 0x2b, 0x65, 0x70, 0x04, 0x22, 0x04, 0x20,
];

/// Each pass request the fake Lys saw: its request line and the audience asked.
type Asked = Arc<Mutex<Vec<(String, String)>>>;

/// A fake Lys agent pass route: each pass is signed by the test Lys for the
/// agent the path names and the audience the body asks.
fn fake_lys() -> Result<(SocketAddr, Asked), Box<dyn Error>> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let address = listener.local_addr()?;
    let asked: Asked = Arc::new(Mutex::new(Vec::new()));
    let log = Arc::clone(&asked);
    std::thread::spawn(move || {
        for stream in listener.incoming() {
            let Ok(mut stream) = stream else { continue };
            let Ok(request) = read_request(&mut stream) else {
                continue;
            };
            let line = request.lines().next().unwrap_or_default().to_owned();
            // The ticket's start probe (AGT-006 R3) connects and sends nothing.
            if line.is_empty() {
                continue;
            }
            let audience = request
                .split("\r\n\r\n")
                .nth(1)
                .and_then(|body| serde_json::from_str::<Value>(body).ok())
                .and_then(|body| body["audience"].as_str().map(str::to_owned))
                .unwrap_or_default();
            let agent = line.split('/').nth(3).unwrap_or_default().to_owned();
            if let Ok(mut log) = log.lock() {
                log.push((line, audience.clone()));
            }
            let rights: &[(&str, &str)] = if audience == "liminal" {
                &[("console.topology", "subscribe"), (SUBJECT, "call")]
            } else {
                &[("haematite.database:studio", "write")]
            };
            let expires_at = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map_or(0, |now| now.as_secs())
                .saturating_add(600);
            let body = lys_pass_aud(&agent, &audience, rights).map_or_else(
                |_| json!({"refusal": "fixture_failed"}),
                |pass| json!({"pass": pass, "expires_at": expires_at, "audience": audience}),
            );
            let body = body.to_string();
            let _sent = write!(
                stream,
                "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                body.len()
            );
        }
    });
    Ok((address, asked))
}

/// The ticket on a `[lys]` table for [`AGENT`] with a fresh key.
fn lys_ticket(
    dir: &std::path::Path,
    server: SocketAddr,
    lys: SocketAddr,
) -> Result<Ticket, Box<dyn Error>> {
    let mut der = PKCS8_ED25519_PREFIX.to_vec();
    der.extend_from_slice(&[7_u8; 32]);
    let pem = format!(
        "-----BEGIN PRIVATE KEY-----\n{}\n-----END PRIVATE KEY-----\n",
        base64::engine::general_purpose::STANDARD.encode(&der)
    );
    let key = dir.join("agent.pem");
    std::fs::write(&key, pem)?;
    let config = dir.join("ticket.toml");
    std::fs::write(
        &config,
        format!(
            "server = \"{server}\"\nchannel = \"{}\"\n\n[lys]\nissuer = \"http://{lys}\"\nagent = \"{AGENT}\"\nkey_file = \"{}\"\n\n[limits]\nframe_bytes = 65536\nrecord_bytes = 1048576\nreply_timeout_ms = 10000\nflow_steps = 8\nrequest_bytes = 65536\n",
            crate::CHANNEL,
            key.display()
        ),
    )?;
    Ticket::spawn(&config)
}

/// A pass's claims, read from its payload segment.
fn claims(pass: &str) -> Result<Value, Box<dyn Error>> {
    let payload = pass.split('.').nth(1).ok_or("the pass has no payload")?;
    let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(payload)?;
    Ok(serde_json::from_slice(&bytes)?)
}

#[test]
fn a_call_with_pass_for_carries_the_callers_own_pass_for_that_audience()
-> Result<(), Box<dyn Error>> {
    let persistence_directory = test_persistence::directory()?;
    // The exchange holds no agent station: the ticket's own inbox is refused
    // in the server's words and its tools keep serving.
    let server = RunningServer::start(persistence_directory.path())?;
    let (lys, asked) = fake_lys()?;
    let (worker, seen) = answering_worker(server.address, 3)?;
    let dir = tempfile::tempdir()?;
    let mut host = Host::open(lys_ticket(dir.path(), server.address, lys)?, true)?;

    let answer = host.tool(
        "call",
        &json!({ "subject": SUBJECT, "input": { "text": "hi" }, "pass_for": AUDIENCE }),
    )?;
    assert_eq!(answer["isError"], false, "{answer}");
    let body = seen.recv_timeout(Duration::from_secs(5))?;
    let pass = body["pass"].as_str().ok_or("the body carries no pass")?;
    let carried = claims(pass)?;
    assert_eq!(carried["aud"], AUDIENCE);
    assert_eq!(carried["sub"], AGENT);
    assert_eq!(carried["holder"]["id"], AGENT);
    assert_eq!(body["text"], "hi");

    // Without pass_for nothing is put in; an input already carrying a pass is refused unsent.
    host.tool(
        "call",
        &json!({ "subject": SUBJECT, "input": { "text": "plain" } }),
    )?;
    let plain = seen.recv_timeout(Duration::from_secs(5))?;
    assert!(plain.get("pass").is_none(), "{plain}");
    let refused = host.tool(
        "call",
        &json!({ "subject": SUBJECT, "input": { "pass": "mine" }, "pass_for": AUDIENCE }),
    )?;
    assert_eq!(refused["isError"], true, "{refused}");

    // A flow step names pass_for the same way.
    host.tool(
        "flow",
        &json!({ "steps": [{ "subject": SUBJECT, "input": { "text": "step" }, "pass_for": AUDIENCE }] }),
    )?;
    let stepped = seen.recv_timeout(Duration::from_secs(5))?;
    assert_eq!(
        claims(stepped["pass"].as_str().ok_or("no pass in the step")?)?["aud"],
        AUDIENCE
    );

    // Lys was asked as the calling agent, for liminal (the ticket's own) and the audience named,
    // and the haematite pass was held and reused rather than asked again.
    let asked = asked.lock().map_err(|_| "poisoned")?.clone();
    assert!(
        asked
            .iter()
            .all(|(line, _)| line.contains(&format!("/api/agents/{AGENT}/pass")))
    );
    let audiences: Vec<&str> = asked
        .iter()
        .map(|(_, audience)| audience.as_str())
        .collect();
    assert_eq!(
        audiences.iter().filter(|a| **a == AUDIENCE).count(),
        1,
        "{audiences:?}"
    );
    assert!(audiences.contains(&"liminal"));
    drop(worker);
    Ok(())
}