liminal-cli 0.2.1

//! The ticket e2e's network: a liminal server started by its own start
//! path, and the test Lys whose key set it admits passes against.
use std::error::Error;
use std::io::Write as _;
use std::net::SocketAddr;
use std::time::Duration;

use ed25519_dalek::{Signer as _, SigningKey};
use lys_pass::{Claims, Holder, Mode, Right};
use serde_json::json;

use crate::{CHANNEL, SUBJECT, reserve_loopback_port};

/// The test Lys: its issuer, the network liminal's subjects sit under, the
/// key it signs passes with, and the bearer a worker connects on.
const ISSUER: &str = "https://lys.test";
const NETWORK: &str = "network-under-test";
const KEY_ID: &str = "ticket-test";
const SEED: [u8; 32] = [11; 32];
pub const BEARER: &str = "bearer-under-test";

/// A liminal server started by its own start path (`runtime::run`), so the
/// console streams the ticket reads are served exactly as in production:
/// a participant configured, `[auth]` with a bearer for the worker, and
/// `[auth.lys]` admitting the test Lys's passes.
pub struct RunningServer {
    pub address: SocketAddr,
    health: SocketAddr,
}

impl RunningServer {
    pub fn start(dir: &std::path::Path) -> Result<Self, Box<dyn Error>> {
        Self::start_with(dir, "")
    }

    /// As [`Self::start`], with `lys_extra` written inside `[auth.lys]`
    /// after its keys: more keys, then any `[[auth.lys.templates]]`.
    pub fn start_with(dir: &std::path::Path, lys_extra: &str) -> Result<Self, Box<dyn Error>> {
        let address = reserve_loopback_port()?;
        let health = reserve_loopback_port()?;
        let keys = lys_keys(dir)?;
        let store = dir.join("store");
        std::fs::create_dir_all(&store)?;
        let config = dir.join("liminal.toml");
        std::fs::write(
            &config,
            format!(
                "listen_address = \"{address}\"\nhealth_listen_address = \"{health}\"\n\
                 drain_timeout_ms = 5000\nrouting_rules = []\npersistence_path = \"{}\"\n\n\
                 [[channels]]\nname = \"{CHANNEL}\"\ndurable = false\n\n\
                 [auth]\ntoken = \"{BEARER}\"\n\n\
                 [auth.lys]\nissuer = \"{ISSUER}\"\nnetwork_id = \"{NETWORK}\"\nkeys_path = \"{}\"\n{lys_extra}\n\n\
                 {PARTICIPANT}",
                store.display(),
                keys.display()
            ),
        )?;
        let (failed, failure) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            if let Err(error) = liminal_server::server::runtime::run(&config) {
                // A test that already finished has dropped the receiver.
                failed.send(error.to_string()).unwrap_or_default();
            }
        });
        wait_ready(health, &failure)?;
        Ok(Self { address, health })
    }

    /// The server's own count of live connections, read from `/metrics`.
    pub fn active_connections(&self) -> Result<u64, Box<dyn Error>> {
        let (_, body) = http_get(self.health, "/metrics")?;
        body.lines()
            .find_map(|line| line.strip_prefix("liminal_connections_active "))
            .ok_or("no liminal_connections_active line")?
            .trim()
            .parse::<u64>()
            .map_err(Into::into)
    }

    /// The live-connection count once it reads `expected`, or the last count
    /// read when [`SETTLE_WITHIN`] passes first. The ticket closes its side
    /// before it answers; the server counts the close when its connection task
    /// reaps it, which under load is after the answer, so a count read once
    /// can be one high for an instant without anything being held.
    pub fn connections_settled_to(&self, expected: u64) -> Result<u64, Box<dyn Error>> {
        let deadline = std::time::Instant::now() + SETTLE_WITHIN;
        loop {
            let now = self.active_connections()?;
            if now == expected || std::time::Instant::now() >= deadline {
                return Ok(now);
            }
            std::thread::yield_now();
        }
    }
}

/// How long a closed connection may take to leave the server's count.
const SETTLE_WITHIN: Duration = Duration::from_secs(5);

/// The participant section of the console harness's live server.
const PARTICIPANT: &str = "[participant]
wire_frame_limit = 65536
attach_receipt_ttl_ms = 60000
receipt_provenance_ttl_ms = 600000
live_receipt_server_report_threshold = 1024
max_live_attach_receipts_per_participant = 8
receipt_provenance_server_report_threshold = 4096
receipt_provenance_per_conversation_report_threshold = 256
max_receipt_provenance_per_participant = 64
max_retired_identity_slots_server = 1024
identity_slots = 4
observer_recovery_max_entries = 64
max_semantic_conversations_per_connection = 32
max_ordinary_record_entries = 1
max_ordinary_record_bytes = 131072
max_generated_marker_entries = 1
max_generated_marker_bytes = 4096
mandatory_transaction_bound_entries = 4
mandatory_transaction_bound_bytes = 16384
full_recovery_claim_entries = 4
full_recovery_claim_bytes = 16384
retained_capacity_entries = 2048
retained_capacity_bytes = 16777216
max_retained_record_rows = 1024
closure_episode_churn_limit = 1024
";

/// One HTTP/1.1 GET on the health endpoint: its status and body.
fn http_get(address: SocketAddr, path: &str) -> Result<(u16, String), Box<dyn Error>> {
    use std::io::Read as _;
    let mut stream = std::net::TcpStream::connect(address)?;
    // One write: the endpoint reads the request in one read and would refuse
    // a request line that arrived in pieces.
    stream.write_all(
        format!("GET {path} HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\n\r\n").as_bytes(),
    )?;
    // The endpoint closes its side once it has written the answer, and a
    // close that races the last read can arrive as a reset: what was read
    // before it is the whole answer.
    let mut bytes = Vec::new();
    let mut chunk = [0_u8; 8192];
    loop {
        match stream.read(&mut chunk) {
            Ok(0) => break,
            Ok(read) => bytes.extend_from_slice(chunk.get(..read).unwrap_or_default()),
            Err(error)
                if error.kind() == std::io::ErrorKind::ConnectionReset && !bytes.is_empty() =>
            {
                break;
            }
            Err(error) => return Err(error.into()),
        }
    }
    let text = String::from_utf8_lossy(&bytes).into_owned();
    let status = text
        .split(' ')
        .nth(1)
        .and_then(|code| code.parse().ok())
        .ok_or("no HTTP status")?;
    let body = text
        .split_once("\r\n\r\n")
        .map(|(_, body)| body.to_owned())
        .unwrap_or_default();
    Ok((status, body))
}

/// The server answers ready on its health endpoint, or the start failed. The
/// server offers no readiness signal to a test, so this is the one bounded
/// wait in the file: it asks again only while the endpoint refuses or answers
/// not ready, and gives up after ten seconds.
fn wait_ready(
    health: SocketAddr,
    failure: &std::sync::mpsc::Receiver<String>,
) -> Result<(), Box<dyn Error>> {
    let deadline = std::time::Instant::now()
        .checked_add(Duration::from_secs(10))
        .ok_or("deadline overflow")?;
    while std::time::Instant::now() < deadline {
        if let Ok(words) = failure.try_recv() {
            return Err(format!("the liminal server did not start: {words}").into());
        }
        if matches!(http_get(health, "/ready"), Ok((200, _))) {
            return Ok(());
        }
        // ast-grep-allow: no-sleep-in-tests: the server gives a test no readiness signal; bounded at ten seconds
        std::thread::sleep(Duration::from_millis(20));
    }
    Err("the liminal server did not become ready within ten seconds".into())
}

/// Unpadded base64url, as a JWT carries its parts.
fn base64url(bytes: &[u8]) -> String {
    const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
    let mut out = String::new();
    for chunk in bytes.chunks(3) {
        let word = chunk.iter().enumerate().fold(0_u32, |word, (index, byte)| {
            word | (u32::from(*byte) << 16_usize.saturating_sub(index.saturating_mul(8)))
        });
        for index in 0..=chunk.len() {
            let shift = 18_usize.saturating_sub(index.saturating_mul(6));
            let sextet = usize::try_from((word >> shift) & 63).unwrap_or_default();
            out.push(char::from(ALPHABET.get(sextet).copied().unwrap_or(b'A')));
        }
    }
    out
}

/// The test Lys's published key set, written where `[auth.lys]` reads it.
fn lys_keys(dir: &std::path::Path) -> Result<std::path::PathBuf, Box<dyn Error>> {
    let key = SigningKey::from_bytes(&SEED);
    let path = dir.join("lys-keys.json");
    let keys = json!({"keys": [{
        "kty": "OKP", "crv": "Ed25519", "kid": KEY_ID, "alg": "EdDSA", "use": "sig",
        "x": base64url(&key.verifying_key().to_bytes()),
    }]});
    std::fs::write(&path, keys.to_string())?;
    Ok(path)
}

/// An agent's liminal pass from the test Lys, holding subscribe on the
/// topology subject and call on the worker's subject, and nothing else.
pub fn lys_pass() -> Result<String, Box<dyn Error>> {
    lys_pass_with(&[("console.topology", "subscribe"), (SUBJECT, "call")])
}

/// An agent's liminal pass holding exactly `rights`, each (subject, action).
pub fn lys_pass_with(rights: &[(&str, &str)]) -> Result<String, Box<dyn Error>> {
    lys_pass_for("agent-under-test", rights)
}

/// `holder`'s liminal pass holding exactly `rights`.
pub fn lys_pass_for(holder: &str, rights: &[(&str, &str)]) -> Result<String, Box<dyn Error>> {
    lys_pass_aud(holder, "liminal", rights)
}

/// `holder`'s pass for `audience` holding exactly `rights`, signed by the test Lys.
pub fn lys_pass_aud(
    holder: &str,
    audience: &str,
    rights: &[(&str, &str)],
) -> Result<String, Box<dyn Error>> {
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)?
        .as_secs();
    let right = |id: &str, action: &str| Right {
        resource: lys_pass::rights::Resource {
            kind: "liminal.subject".to_owned(),
            id: id.to_owned(),
        },
        actions: vec![action.to_owned()],
        mode: Mode::Outright,
        grant: format!("grant-{action}"),
    };
    let claims = Claims {
        iss: ISSUER.to_owned(),
        sub: holder.to_owned(),
        aud: audience.to_owned(),
        iat: now.saturating_sub(60),
        exp: now.saturating_add(600),
        nbf: None,
        holder: Holder {
            id: holder.to_owned(),
            kind: "agent".to_owned(),
            responsible: Some("person-under-test".to_owned()),
        },
        agent_key_thumbprints: Vec::new(),
        rights: rights
            .iter()
            .map(|(subject, action)| right(subject, action))
            .collect(),
        rights_truncated: None,
    };
    let header = base64url(
        json!({"alg": "EdDSA", "typ": "JWT", "kid": KEY_ID})
            .to_string()
            .as_bytes(),
    );
    let payload = base64url(&serde_json::to_vec(&claims)?);
    let signature = SigningKey::from_bytes(&SEED).sign(format!("{header}.{payload}").as_bytes());
    Ok(format!(
        "{header}.{payload}.{}",
        base64url(&signature.to_bytes())
    ))
}