liminal-cli 0.3.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,
    /// The `[websocket]` door, path `/ws` (AGT-006 R2).
    pub websocket: 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 websocket = 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\
                 [websocket]\nlisten_address = \"{websocket}\"\npath = \"/ws\"\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,
            websocket,
            health,
        })
    }

    /// The ticket's seat address for the WebSocket door.
    pub fn websocket_url(&self) -> String {
        format!("ws://{}/ws", self.websocket)
    }

    /// 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();
        }
    }
}

/// An exchange with no `[auth]` (ACCESS-010 R5), started as a child of the
/// liminal binary so its log can be read: no keys, no bearer, no Lys. The
/// child is stopped when the server is dropped.
pub struct OpenServer {
    pub address: SocketAddr,
    pub health: SocketAddr,
    child: Option<std::process::Child>,
    log: std::path::PathBuf,
}

impl OpenServer {
    /// The configuration of an open exchange in `dir`, listening on `host`
    /// (loopback, or an address R1 refuses) at reserved ports.
    pub fn write_config(
        dir: &std::path::Path,
        host: &str,
    ) -> Result<std::path::PathBuf, Box<dyn Error>> {
        let address = reserve_loopback_port()?;
        let health = reserve_loopback_port()?;
        let store = dir.join("store");
        std::fs::create_dir_all(&store)?;
        let config = dir.join("liminal.toml");
        std::fs::write(
            &config,
            format!(
                "listen_address = \"{host}:{}\"\nhealth_listen_address = \"{health}\"\n\
                 drain_timeout_ms = 5000\nrouting_rules = []\npersistence_path = \"{}\"\n\n\
                 [[channels]]\nname = \"{CHANNEL}\"\ndurable = false\n\n{PARTICIPANT}",
                address.port(),
                store.display(),
            ),
        )?;
        Ok(config)
    }

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

    /// The exchange written by [`Self::write_config`] on `host`, started as a
    /// child: ready on its health endpoint, or the error names the exit and
    /// the child's stderr (an address R1 refuses never reaches ready).
    pub fn start_on(dir: &std::path::Path, host: &str) -> Result<Self, Box<dyn Error>> {
        let config = Self::write_config(dir, host)?;
        let text = std::fs::read_to_string(&config)?;
        let field = |name: &str| -> Result<SocketAddr, Box<dyn Error>> {
            let line = text
                .lines()
                .find(|line| line.starts_with(name))
                .ok_or_else(|| format!("no {name} line"))?;
            Ok(line
                .trim_start_matches(name)
                .trim_matches(|c| c == ' ' || c == '=' || c == '"')
                .parse()?)
        };
        let address = field("listen_address")?;
        let health = field("health_listen_address")?;
        let log = dir.join("server.log");
        let child = std::process::Command::new(env!("CARGO_BIN_EXE_liminal"))
            .arg("--config")
            .arg(&config)
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::null())
            .stderr(std::fs::File::create(&log)?)
            .spawn()?;
        let mut server = Self {
            address,
            health,
            child: Some(child),
            log,
        };
        server.wait_ready()?;
        Ok(server)
    }

    /// The child's log so far (its stderr).
    pub fn log(&self) -> Result<String, Box<dyn Error>> {
        Ok(std::fs::read_to_string(&self.log)?)
    }

    /// Ready on the health endpoint, or the child exited: the one bounded
    /// wait, as [`wait_ready`] is for the in-process server.
    fn wait_ready(&mut self) -> 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 Some(child) = self.child.as_mut()
                && let Some(status) = child.try_wait()?
            {
                return Err(format!(
                    "the open exchange did not start ({status}): {}",
                    self.log()?
                )
                .into());
            }
            if matches!(http_get(self.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 open exchange did not become ready within ten seconds".into())
    }
}

impl Drop for OpenServer {
    fn drop(&mut self) {
        if let Some(mut child) = self.child.take() {
            drop(child.kill());
            drop(child.wait());
        }
    }
}

/// 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())
    ))
}