liminal-server 0.21.2

Standalone server for the liminal messaging bus
Documentation
use crate::ServerError;
use crate::config::types::ServerConfig;
use crate::link_handshake::LocalMachine;
use lys_core::keys::Ed25519Identity;
use zeroize::Zeroizing;

type TestResult = Result<(), Box<dyn std::error::Error>>;

fn configured(
    exports: &[&str],
    channels: &[(&str, bool)],
) -> Result<ServerConfig, toml::de::Error> {
    toml::from_str(&configuration_text(exports, channels))
}

fn configuration_text(exports: &[&str], channels: &[(&str, bool)]) -> String {
    let peer = LocalMachine::new(Ed25519Identity::from_seed(&Zeroizing::new([2; 32])));
    let channels = if channels.is_empty() {
        "channels = []\n".to_owned()
    } else {
        channels
            .iter()
            .map(|(name, durable)| format!("[[channels]]\nname = {name:?}\ndurable = {durable}\n"))
            .collect::<String>()
    };
    format!(
        r#"
listen_address = "127.0.0.1:8080"
health_listen_address = "127.0.0.1:8081"
drain_timeout_ms = 0
routing_rules = []
persistence_path = "/private/tmp"
{channels}
[auth]
token = "test-token"
[auth.lys]
issuer = "https://issuer.example"
network_id = "network-a"
keys_path = "issuer.json"
[auth.lys.machine]
identity_path = "identity.key"
pass_path = "machine.pass"
[auth.lys.machine.machines]
peer = "{}"
[[auth.lys.machine.links]]
link_id = "named-link"
peer_holder = "peer"
exports = {exports:?}
"#,
        hex::encode(peer.machine().ed25519_public())
    )
}

fn named_refusal(result: Result<(), ServerError>, prefix: &str, cause: &str) {
    let Err(ServerError::ConfigValidation { message }) = result else {
        panic!("expected named ConfigValidation for {prefix}: {result:?}");
    };
    assert!(message.contains("named-link"), "{message}");
    assert!(
        message.contains(&format!("exports '{prefix}'")),
        "{message}"
    );
    assert!(message.contains(cause), "{message}");
}

fn refusal(config: &mut ServerConfig, prefix: &str, cause: &str) -> TestResult {
    named_refusal(super::validate(config, None), prefix, cause);
    let link = config
        .auth
        .as_ref()
        .and_then(|auth| auth.lys.as_ref())
        .and_then(|lys| lys.machine.as_ref())
        .and_then(|machine| machine.links.first())
        .ok_or("configured link absent")?;
    let exports = link.exports.iter().map(String::as_str).collect::<Vec<_>>();
    let channels = config
        .channels
        .iter()
        .map(|channel| (channel.name.as_str(), channel.durable))
        .collect::<Vec<_>>();
    let root = tempfile::tempdir()?;
    let path = root.path().join("liminal.toml");
    std::fs::write(&path, configuration_text(&exports, &channels))?;
    named_refusal(
        crate::config::file::load_config(&path).map(|_| ()),
        prefix,
        cause,
    );
    Ok(())
}

#[test]
fn link_export_without_a_placed_subject_is_refused_by_name() -> TestResult {
    let mut config = configured(&["embed", "demo"], &[("embed.result", true)])?;
    refusal(&mut config, "demo", "names no placed subject")?;
    let mut config = configured(&["orders"], &[("orders_other", true)])?;
    refusal(&mut config, "orders", "names no placed subject")?;
    Ok(())
}

#[test]
fn link_export_of_a_non_durable_channel_is_refused_by_name() -> TestResult {
    let mut config = configured(&["console"], &[("console.topology", false)])?;
    refusal(
        &mut config,
        "console",
        "non-durable channel 'console.topology'",
    )?;
    let mut config = configured(
        &["orders"],
        &[("orders.created", true), ("orders.live", false)],
    )?;
    refusal(&mut config, "orders", "non-durable channel 'orders.live'")?;
    Ok(())
}

#[test]
fn link_export_of_the_reserved_namespace_is_refused_by_name() -> TestResult {
    for prefix in ["liminal", "liminal.", "liminal.cost", "liminal.cost.child"] {
        let mut config = configured(&[prefix], &[("liminal.cost", true)])?;
        refusal(&mut config, prefix, "reserved 'liminal.' namespace")?;
    }
    Ok(())
}

#[test]
fn link_export_of_durable_placed_subjects_passes_validation() -> TestResult {
    let mut config = configured(
        &["orders", "liminal_other"],
        &[
            ("orders.created", true),
            ("liminal_other", true),
            ("orders_other", false),
        ],
    )?;
    super::validate(&mut config, None)?;
    Ok(())
}

fn service_configuration(exports: &[&str], channels: &[(&str, bool)], section: &str) -> String {
    let text = configuration_text(exports, channels).replace(
        "keys_path = \"issuer.json\"",
        "keys_path = \"issuer.json\"\napp_credential_path = \"app.credential\"",
    );
    format!("{text}\n{section}")
}

fn validate_service_text(text: &str, prefix: &str, subject: &str) -> TestResult {
    let mut config: ServerConfig = toml::from_str(text)?;
    let cause = format!("names service subject '{subject}', which is not a stream");
    named_refusal(super::validate(&mut config, None), prefix, &cause);
    let root = tempfile::tempdir()?;
    let path = root.path().join("liminal.toml");
    std::fs::write(&path, text)?;
    named_refusal(
        crate::config::file::load_config(&path).map(|_| ()),
        prefix,
        &cause,
    );
    Ok(())
}

const EMBED_TEMPLATE: &str = r#"
[[auth.lys.templates]]
name = "embed"
command = "/bin/sh"
subjects = ["embed"]
stop_timeout_ms = 500
"#;

#[test]
fn link_export_of_a_template_service_is_refused_as_not_a_stream() -> TestResult {
    for prefix in ["embed", "embed.bge"] {
        let text = service_configuration(&[prefix], &[], EMBED_TEMPLATE);
        validate_service_text(&text, prefix, prefix)?;
    }
    let mut config: ServerConfig = toml::from_str(&service_configuration(
        &["embed_other", "embed.bge.extra"],
        &[],
        EMBED_TEMPLATE,
    ))?;
    let Err(ServerError::ConfigValidation { message }) = super::validate(&mut config, None) else {
        return Err("unserved subject was accepted".into());
    };
    for prefix in ["embed_other", "embed.bge.extra"] {
        assert!(
            message.contains(&format!("exports '{prefix}' names no placed subject")),
            "{message}"
        );
    }
    Ok(())
}

#[test]
fn link_export_of_a_service_and_durable_channel_exports_the_stream() -> TestResult {
    validate_service_text(
        &service_configuration(&["embed"], &[], EMBED_TEMPLATE),
        "embed",
        "embed",
    )?;
    for channel in ["embed.result", "embed"] {
        let text = service_configuration(&["embed"], &[(channel, true)], EMBED_TEMPLATE);
        let root = tempfile::tempdir()?;
        let path = root.path().join("liminal.toml");
        std::fs::write(&path, text)?;
        crate::config::file::load_config(&path)?;
    }
    Ok(())
}

#[test]
fn link_export_of_a_service_and_non_durable_channel_names_both_causes() -> TestResult {
    let text = service_configuration(&["embed"], &[("embed.live", false)], EMBED_TEMPLATE);
    validate_service_text(&text, "embed", "embed")?;
    let mut config: ServerConfig = toml::from_str(&text)?;
    named_refusal(
        super::validate(&mut config, None),
        "embed",
        "non-durable channel 'embed.live'",
    );
    Ok(())
}

#[test]
fn link_export_of_configured_administrative_services_is_not_a_stream() -> TestResult {
    let section = "[auth.lys.services]\nmachine = 'local'\nhost = 'host'";
    for (prefix, subject) in [
        ("alarms", "alarms"),
        ("seats", "seats.<name>"),
        ("seats.agent", "seats.agent"),
        ("sessions.session", "sessions.session.keys"),
        ("sessions.session.keys", "sessions.session.keys"),
    ] {
        validate_service_text(
            &service_configuration(&[prefix], &[], section),
            prefix,
            subject,
        )?;
    }
    let mut config = configured(&["alarms"], &[])?;
    named_refusal(
        super::validate(&mut config, None),
        "alarms",
        "names no placed subject",
    );
    Ok(())
}

#[test]
fn link_export_of_configured_broker_and_relay_services_is_not_a_stream() -> TestResult {
    for (prefix, section, subject) in [
        (
            "net.broker",
            "[auth.lys.broker]\nudp_listen = '127.0.0.1:8090'",
            "net.broker.register",
        ),
        ("net.relay", "[auth.lys.relay]\nenabled = true", "net.relay"),
    ] {
        validate_service_text(
            &service_configuration(&[prefix], &[], section),
            prefix,
            subject,
        )?;
    }
    let mut config: ServerConfig = toml::from_str(&service_configuration(
        &["net.relay"],
        &[],
        "[auth.lys.relay]\nenabled = false",
    ))?;
    named_refusal(
        super::validate(&mut config, None),
        "net.relay",
        "names no placed subject",
    );
    Ok(())
}

#[test]
fn link_export_of_template_lifecycle_calls_is_not_a_stream() -> TestResult {
    let text = service_configuration(&["services.list"], &[], EMBED_TEMPLATE);
    validate_service_text(&text, "services.list", "services.list")?;
    Ok(())
}