liminal-server 0.25.4

//! The bindings a production configuration builds, driven from the
//! configuration alone: broker-only binds the echo and refuses unadmitted
//! calls, a machine pass is read and verified before any service is bound.

use super::ServiceBindings;
use crate::config::types::LysConfig;
use crate::server::connection::LiminalConnectionServices;
use crate::services::ServiceRequest;
use lys_core::keys::Ed25519Identity;
use serde_json::json;
use std::net::UdpSocket;
use std::sync::Arc;
use std::time::Duration;

fn base64url(bytes: &[u8]) -> String {
    const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
    let mut encoded = String::new();
    for chunk in bytes.chunks(3) {
        let first = chunk[0];
        encoded.push(char::from(ALPHABET[usize::from(first >> 2)]));
        let second = chunk.get(1).copied().unwrap_or(0);
        encoded.push(char::from(
            ALPHABET[usize::from((first & 3) << 4 | second >> 4)],
        ));
        if chunk.len() > 1 {
            let third = chunk.get(2).copied().unwrap_or(0);
            encoded.push(char::from(
                ALPHABET[usize::from((second & 15) << 2 | third >> 6)],
            ));
            if chunk.len() > 2 {
                encoded.push(char::from(ALPHABET[usize::from(third & 63)]));
            }
        }
    }
    encoded
}

fn machine_pass() -> Result<(String, String), serde_json::Error> {
    let issuer = Ed25519Identity::from_seed(&zeroize::Zeroizing::new([9; 32]));
    let keys = json!({"keys": [{"kty": "OKP", "crv": "Ed25519", "kid": "broker",
        "x": base64url(&issuer.public_key_bytes()), "alg": "EdDSA", "use": "sig"}]});
    let claims = json!({"iss": "https://issuer.example", "sub": "broker", "aud": "liminal",
        "iat": 0, "exp": 20, "rights": [],
        "holder": {"id": "broker", "kind": "machine", "responsible": "owner"}});
    let header = base64url(br#"{"alg":"EdDSA","typ":"JWT","kid":"broker"}"#);
    let signed = format!("{header}.{}", base64url(&serde_json::to_vec(&claims)?));
    let token = format!("{signed}.{}", base64url(&issuer.sign(signed.as_bytes())));
    Ok((keys.to_string(), token))
}

#[test]
fn broker_only_configuration_binds_echo_and_refuses_unadmitted_calls()
-> Result<(), Box<dyn std::error::Error>> {
    let root = tempfile::tempdir()?;
    let keys = root.path().join("issuer-keys.json");
    let pass = root.path().join("machine.pass");
    let (keys_json, token) = machine_pass()?;
    std::fs::write(&keys, keys_json)?;
    std::fs::write(&pass, token)?;
    let mut lys: LysConfig = serde_json::from_value(json!({
        "issuer": "https://issuer.example", "network_id": "network-a",
        "keys_path": keys, "broker": {"udp_listen": "127.0.0.1:0"},
        "machine": {"identity_path": "unused", "pass_path": pass}
    }))?;
    let runtime = tokio::runtime::Builder::new_current_thread().build()?;
    let services = Arc::new(LiminalConnectionServices::empty()?);
    let bindings =
        ServiceBindings::configure(&lys, services.clone(), runtime.handle().clone(), 10)?
            .ok_or("broker did not configure")?;
    let refused = bindings
        .serve_broker(
            None,
            "net.broker.lookup",
            &ServiceRequest {
                operation: "lookup",
                payload: b"{}",
            },
            10,
        )
        .err()
        .ok_or("unadmitted call accepted")?;
    assert_eq!(refused.to_string(), "pass_not_admitted");
    let refused = bindings
        .services()
        .seats
        .start("unused", "operation")
        .err()
        .ok_or("unconfigured administrative service acted")?;
    assert!(
        refused
            .to_string()
            .contains("auth.lys.services is not configured")
    );
    let address = bindings.echo_address().ok_or("echo missing")?;
    assert_ne!(address.port(), 0);
    let counters = bindings.broker().ok_or("broker missing")?.counters();
    assert_eq!(counters.snapshot().echo_answered, 0);
    let socket = UdpSocket::bind("127.0.0.1:0")?;
    socket.set_read_timeout(Some(Duration::from_secs(3)))?;
    socket.send_to(&[b"LMOB".as_slice(), &[3; 16]].concat(), address)?;
    let mut reply = [0; 64];
    let (length, _) = socket.recv_from(&mut reply)?;
    assert_eq!(length, 27);
    assert_eq!(&reply[..20], &[b"LMOA".as_slice(), &[3; 16]].concat());
    drop(bindings);
    assert_eq!(counters.snapshot().echo_answered, 1);
    let rebound = UdpSocket::bind(address)?;
    assert_eq!(rebound.local_addr()?, address);
    lys.broker = None;
    lys.services = Some(toml::from_str("machine = 'unused'\nhost = 'broker'")?);
    let bindings = ServiceBindings::configure(&lys, services, runtime.handle().clone(), 10)?
        .ok_or("administrative services did not configure")?;
    assert_eq!(bindings.echo_address(), None);
    Ok(())
}