liminal-server 0.25.4

//! Two socket servers sharing one issuer and mutually recorded machine keys.

use liminal::protocol::{Frame, FrameHeader, MessageEnvelope, ProtocolVersion, SchemaId, codec};
use liminal_protocol::credential::WireCredential;
use liminal_server::auth_lys::{
    GrantMode, LysPassConsumer,
    lys_pass_adapter::{LiveAnswer, LiveLysAsk, LysPassAdapter},
    session::ConsumerAdmitter,
};
use liminal_server::config::types::{MachineConfig, OutboundLinkConfig};
use liminal_server::config::{ChannelDef, LimitsConfig, ServerConfig, ServicesConfig};
use liminal_server::link::LinkHost;
use liminal_server::link_handshake::{LocalMachine, MachineIdentity};
use liminal_server::server::connection::ConnectionSupervisor;
use liminal_server::server::listener::ServerListener;
use lys_core::keys::Ed25519Identity;
use std::collections::BTreeMap;
use std::error::Error;
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpStream};
use std::path::Path;
use std::sync::{
    Arc,
    atomic::{AtomicUsize, Ordering},
};
use std::time::Duration;
use zeroize::Zeroizing;

#[path = "federation/issuer.rs"]
pub mod issuer;
use issuer::{base64url, pass};
#[path = "federation/contract_tests.rs"]
mod contract_tests;
#[path = "federation/rights.rs"]
mod rights;

const ISSUER: &str = "https://federation.example";
const NETWORK: &str = "network-a";
const SUBJECTS: [&str; 3] = ["orders", "orders.created", "orders.cancelled"];
type FixtureResult<T> = Result<T, Box<dyn Error>>;

/// The placed subjects served by each fixture end.
#[must_use]
pub fn configured_channels(exporting: bool) -> Vec<ChannelDef> {
    SUBJECTS
        .iter()
        .map(|name| ChannelDef {
            name: (*name).to_owned(),
            schema_ref: None,
            durable: exporting,
            loaded_schema: None,
        })
        .collect()
}

#[derive(Clone, Copy, Debug)]
pub enum LiveOpen {
    Outright,
    Held,
    Refused,
}

#[derive(Debug)]
struct OpenAuthority {
    answer: LiveOpen,
    asks: AtomicUsize,
    export_right: bool,
}

impl LiveLysAsk for OpenAuthority {
    fn ask(
        &self,
        _pass: &str,
        holder: &str,
        target: &lys_pass::Target,
        _now: u64,
    ) -> Result<LiveAnswer, lys_pass::Error> {
        self.asks.fetch_add(1, Ordering::SeqCst);
        if holder != "exporter"
            || target.kind != "liminal.link"
            || target.id != "orders-link"
            || target.action != "link_export"
        {
            return Ok(LiveAnswer::Refused {
                refusal: "unknown_link_open".to_owned(),
            });
        }
        Ok(
            match if self.export_right {
                self.answer
            } else {
                LiveOpen::Refused
            } {
                LiveOpen::Outright => LiveAnswer::Outright {
                    grant: "grant-link".to_owned(),
                },
                LiveOpen::Held => LiveAnswer::Held {
                    mode: GrantMode::ByDraft,
                    grant: "grant-link".to_owned(),
                    draft: "draft-link".to_owned(),
                },
                LiveOpen::Refused => LiveAnswer::Refused {
                    refusal: "link_export_refused".to_owned(),
                },
            },
        )
    }
}

#[must_use]
pub fn local_machine(seed: u8) -> LocalMachine {
    LocalMachine::new(Ed25519Identity::from_seed(&Zeroizing::new([seed; 32])))
}

pub struct Federation {
    pub exporter: FederationServer,
    pub importer: FederationServer,
    authority: Arc<OpenAuthority>,
    root: tempfile::TempDir,
}

impl Federation {
    /// Starts two mutually admitted socket servers with the requested open answer.
    ///
    /// # Errors
    /// Temporary storage, pass encoding or verification, server setup, or binding failed.
    pub fn start(answer: LiveOpen) -> FixtureResult<Self> {
        Self::with_rights(answer, true, true)
    }

    /// Starts two servers with independently granted link ends.
    /// # Errors
    /// Pass encoding, verification, storage, or listener setup failed.
    pub fn with_rights(answer: LiveOpen, export: bool, import: bool) -> FixtureResult<Self> {
        Self::build(answer, export, import, false)
    }

    /// Refuses one importing decision, then uses the verified importing pass.
    /// # Errors
    /// Pass encoding, verification, storage, or listener setup failed.
    pub fn with_first_import_refused() -> FixtureResult<Self> {
        Self::build(LiveOpen::Outright, true, true, true)
    }

    fn build(
        answer: LiveOpen,
        export: bool,
        import: bool,
        first_import_refused: bool,
    ) -> FixtureResult<Self> {
        let root = tempfile::Builder::new()
            .prefix("liminal-federation-")
            .tempdir()?;
        let issuer = Ed25519Identity::from_seed(&Zeroizing::new([9; 32]));
        let keys = serde_json::json!({ "keys": [{ "kty": "OKP", "crv": "Ed25519", "kid": "federation", "x": base64url(&issuer.public_key_bytes()), "alg": "EdDSA", "use": "sig" }] });
        let authority = Arc::new(OpenAuthority {
            answer,
            asks: AtomicUsize::new(0),
            export_right: export,
        });
        let adapter = Arc::new(
            LysPassAdapter::from_keys_json(&serde_json::to_string(&keys)?, ISSUER.to_owned())?
                .with_live(authority.clone()),
        );
        let mut exporter = FederationServer::start(
            root.path(),
            &issuer,
            &adapter,
            (1, "exporter", if export { "link_export" } else { "" }),
            ("importer", local_machine(2).machine()),
            false,
        )?;
        let importer = FederationServer::start(
            root.path(),
            &issuer,
            &adapter,
            (2, "importer", if import { "link_import" } else { "" }),
            ("exporter", exporter.identity()),
            first_import_refused,
        )?;
        exporter.config.links.push(OutboundLinkConfig {
            link_id: "orders-link".to_owned(),
            peer_holder: "importer".to_owned(),
            peer_address: Some(importer.address.to_string()),
            exports: vec!["orders".to_owned()],
        });
        Ok(Self {
            exporter,
            importer,
            authority,
            root,
        })
    }

    #[must_use]
    pub fn outbound(&self) -> OutboundLinkConfig {
        self.exporter.config.links[0].clone()
    }

    #[must_use]
    pub fn live_asks(&self) -> usize {
        self.authority.asks.load(Ordering::SeqCst)
    }

    /// Stops both servers and removes their temporary storage.
    ///
    /// # Errors
    /// Listener shutdown or removal of temporary storage failed.
    pub fn shutdown(mut self) -> FixtureResult<()> {
        self.exporter.shutdown()?;
        self.importer.shutdown()?;
        let Self {
            exporter,
            importer,
            authority: _,
            root,
        } = self;
        drop(exporter);
        drop(importer);
        root.close()?;
        Ok(())
    }
}

pub struct FederationServer {
    pub address: SocketAddr,
    pub host: Arc<LinkHost>,
    pub config: MachineConfig,
    pub supervisor: ConnectionSupervisor,
    listener: Option<ServerListener>,
}

impl FederationServer {
    fn start(
        root: &Path,
        issuer: &Ed25519Identity,
        adapter: &Arc<LysPassAdapter>,
        local: (u8, &str, &str),
        remote: (&str, MachineIdentity),
        first_import_refused: bool,
    ) -> FixtureResult<Self> {
        let (seed, holder, action) = local;
        let (peer_holder, peer) = remote;
        let directory = root.join(holder);
        std::fs::create_dir(&directory)?;
        let identity_path = directory.join("machine.key");
        let pass_path = directory.join("machine.pass");
        std::fs::write(&identity_path, hex::encode([seed; 32]))?;
        let token = pass(issuer, holder, action)?;
        std::fs::write(&pass_path, &token)?;
        let identity_path = toml::Value::String(
            identity_path
                .to_str()
                .ok_or("fixture identity path is not UTF-8")?
                .to_owned(),
        );
        let pass_path = toml::Value::String(
            pass_path
                .to_str()
                .ok_or("fixture pass path is not UTF-8")?
                .to_owned(),
        );
        let mut machine: MachineConfig = toml::from_str(&format!(
            "identity_path = {identity_path}\npass_path = {pass_path}\n"
        ))?;
        machine
            .machines
            .insert(peer_holder.to_owned(), hex::encode(peer.ed25519_public()));
        let verified = adapter.verify_offline(token.as_bytes(), "liminal", 0)?;
        let host = Arc::new(LinkHost::new(
            local_machine(seed),
            Box::new(rights::Rights::new(adapter, verified, first_import_refused)),
            BTreeMap::from([(peer_holder.to_owned(), peer)]),
        ));
        let config = ServerConfig {
            listen_address: "127.0.0.1:0".parse()?,
            extra_listen_addresses: Vec::new(),
            health_listen_address: "127.0.0.1:0".parse()?,
            channels: configured_channels(holder == "exporter"),
            routing_rules: Vec::new(),
            persistence_path: Some(directory),
            cluster: None,
            auth: None,
            drain_timeout_ms: 0,
            services: ServicesConfig::default(),
            limits: LimitsConfig::default(),
            participant: None,
            websocket: None,
        };
        let supervisor = ConnectionSupervisor::from_config(&config)?;
        supervisor.install_lys_for_tests(
            Arc::new(ConsumerAdmitter::new(Arc::clone(adapter))),
            NETWORK.to_owned(),
            Arc::clone(&host),
        )?;
        let listener = ServerListener::bind(&config, supervisor.clone())?;
        Ok(Self {
            address: listener.local_addr(),
            host,
            config: machine,
            supervisor,
            listener: Some(listener),
        })
    }

    #[must_use]
    pub fn identity(&self) -> MachineIdentity {
        self.host.local().machine()
    }

    /// Connects and admits the machine pass from the supplied file.
    ///
    /// # Errors
    /// The pass could not be read, socket I/O or framing failed, or admission was refused.
    pub fn connect_as(&self, pass_path: &Path) -> FixtureResult<TcpStream> {
        let mut socket = TcpStream::connect(self.address)?;
        socket.set_nodelay(true)?;
        // Socket answers are signals; this bounds only a failed observation.
        socket.set_read_timeout(Some(Duration::from_secs(5)))?;
        let token = Zeroizing::new(std::fs::read_to_string(pass_path)?);
        write_frame(
            &mut socket,
            &Frame::Connect {
                flags: 0,
                min_version: ProtocolVersion::new(1, 0),
                max_version: ProtocolVersion::new(1, 0),
                auth_token: WireCredential::lys_record(token.trim().as_bytes()),
            },
        )?;
        if !matches!(read_frame(&mut socket)?, Frame::ConnectAck { .. }) {
            return Err("machine pass was not admitted".into());
        }
        Ok(socket)
    }

    /// Opens an admitted local subscription to the subject.
    ///
    /// # Errors
    /// Connection, socket I/O or framing failed, or subscription was refused.
    pub fn subscribe(&self, subject: &str) -> FixtureResult<LocalSubscriber> {
        let mut socket = self.connect_as(&self.config.pass_path)?;
        write_frame(
            &mut socket,
            &Frame::Subscribe {
                flags: 0,
                stream_id: 1,
                channel: subject.to_owned(),
                accepted_schemas: Vec::new(),
                max_in_flight: 32,
            },
        )?;
        let Frame::SubscribeAck {
            stream_id: 1,
            selected_schema,
            ..
        } = read_frame(&mut socket)?
        else {
            return Err("local subscription was not admitted".into());
        };
        Ok(LocalSubscriber {
            socket,
            selected_schema,
        })
    }

    /// Publishes one envelope and waits for its protocol answer.
    ///
    /// # Errors
    /// Connection, socket I/O or framing failed, or publishing was refused.
    pub fn publish(&self, subject: &str, envelope: MessageEnvelope) -> FixtureResult<()> {
        let mut socket = self.connect_as(&self.config.pass_path)?;
        write_frame(
            &mut socket,
            &Frame::Publish {
                flags: 0,
                stream_id: 1,
                channel: subject.to_owned(),
                envelope,
                idempotency_key: None,
            },
        )?;
        match read_frame(&mut socket)? {
            Frame::PublishAck { stream_id: 1, .. } => Ok(()),
            Frame::PublishError {
                stream_id,
                reason_code,
                message,
                ..
            } => Err(format!(
                "local publish was not admitted: PublishError stream_id={stream_id} code={reason_code:#06x} reason={message:?}"
            )
            .into()),
            frame => Err(format!(
                "local publish was not admitted: frame={:?}",
                frame.frame_type()
            )
            .into()),
        }
    }

    fn shutdown(&mut self) -> FixtureResult<()> {
        if let Some(listener) = self.listener.take() {
            listener.shutdown()?;
            self.supervisor.force_close_active_connections();
            self.supervisor.scheduler().shutdown();
        }
        Ok(())
    }
}

impl Drop for FederationServer {
    fn drop(&mut self) {
        if let Err(error) = self.shutdown() {
            eprintln!("federation server cleanup failed: {error}");
        }
    }
}

pub struct LocalSubscriber {
    socket: TcpStream,
    selected_schema: SchemaId,
}

impl LocalSubscriber {
    #[must_use]
    pub const fn selected_schema(&self) -> SchemaId {
        self.selected_schema
    }

    /// Reads one delivery from the admitted subscription.
    ///
    /// # Errors
    /// Socket I/O or decoding failed, or the peer answered with a different frame.
    pub fn receive(&mut self) -> FixtureResult<MessageEnvelope> {
        match read_frame(&mut self.socket)? {
            Frame::Deliver {
                stream_id: 1,
                envelope,
                ..
            } => Ok(envelope),
            _ => Err("local subscriber received no delivery".into()),
        }
    }
}

/// Encodes and writes one protocol frame.
///
/// # Errors
/// Frame sizing, encoding, or socket I/O failed.
pub fn write_frame(socket: &mut TcpStream, frame: &Frame) -> FixtureResult<()> {
    let mut bytes = vec![0; codec::encoded_len(frame)?];
    let length = codec::encode(frame, &mut bytes)?;
    socket.write_all(&bytes[..length])?;
    Ok(())
}

/// Reads one frame within the fixture's inbox bound.
///
/// # Errors
/// Socket I/O, length conversion or decoding failed, or the frame exceeds the bound.
pub fn read_frame(socket: &mut TcpStream) -> FixtureResult<Frame> {
    let mut header = [0; FrameHeader::WIRE_LEN];
    socket.read_exact(&mut header)?;
    let length = usize::try_from(u32::from_be_bytes([
        header[6], header[7], header[8], header[9],
    ]))?;
    if length > LimitsConfig::DEFAULT_MAX_CONNECTION_INBOX_BYTES {
        return Err("fixture peer frame exceeds the inbox bound".into());
    }
    let mut bytes = vec![0; header.len() + length];
    bytes[..header.len()].copy_from_slice(&header);
    socket.read_exact(&mut bytes[header.len()..])?;
    Ok(codec::decode(&bytes)?.0)
}