liminal-server 0.25.0

//! A blind broker: address registrations, lookups, punch instructions and a UDP echo.

use super::keys::{ServicePublisher, service_envelope};
use super::{ServiceEffectError, ServiceRequest};
use crate::auth_lys::session::LysSession;
use crate::auth_lys::{AccessDecisionError, HotAction, SubjectBinding};
use crate::config::types::LysConfig;
use crate::link_handshake::MachineIdentity;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::collections::BTreeMap;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, UdpSocket};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;

/// A control call refused before its effect, or a failed punch publication.
#[derive(Debug, thiserror::Error)]
pub enum BrokerRefusal {
    /// The caller's pass did not grant this call.
    #[error("{0}")]
    Decision(#[from] AccessDecisionError),
    /// A named boundary or state refusal.
    #[error("{0}")]
    Named(&'static str),
    /// The publisher refused the instruction.
    #[error("{0}")]
    Publish(#[from] ServiceEffectError),
}

/// Traffic counters; there is deliberately no mutable data-byte counter.
#[derive(Debug, Default)]
pub struct BrokerCounters {
    registered: AtomicU64,
    lookups: AtomicU64,
    punches_told: AtomicU64,
    echo_answered: AtomicU64,
    echo_dropped: AtomicU64,
}

/// Successful registrations and instructions, lookup attempts and echo traffic.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub struct CounterSnapshot {
    /// Successful registrations, including replacements.
    pub registered: u64,
    /// Authorized, well-formed lookup attempts.
    pub lookups: u64,
    /// Instructions accepted by the publisher.
    pub punches_told: u64,
    /// Echo replies sent.
    pub echo_answered: u64,
    /// Datagrams rejected by the echo protocol.
    pub echo_dropped: u64,
    /// The broker has no data-forwarding operation, so this is always zero.
    pub data_bytes_forwarded: u64,
}

impl BrokerCounters {
    /// Reads each traffic counter; data bytes are pinned to zero.
    #[must_use]
    pub fn snapshot(&self) -> CounterSnapshot {
        CounterSnapshot {
            registered: self.registered.load(Ordering::Relaxed),
            lookups: self.lookups.load(Ordering::Relaxed),
            punches_told: self.punches_told.load(Ordering::Relaxed),
            echo_answered: self.echo_answered.load(Ordering::Relaxed),
            echo_dropped: self.echo_dropped.load(Ordering::Relaxed),
            data_bytes_forwarded: 0,
        }
    }
}

#[derive(Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Registration {
    device: String,
    addresses: Vec<String>,
    #[serde(default)]
    observed: Option<SocketAddr>,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Lookup {
    device: String,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Punch {
    to: String,
    nonce: String,
}

/// In-memory control state, resolved from the caller's verified holder.
pub struct Broker {
    network: String,
    recorded: BTreeMap<String, MachineIdentity>,
    registrations: Mutex<BTreeMap<[u8; 32], Registration>>,
    publisher: Arc<dyn ServicePublisher>,
    counters: Arc<BrokerCounters>,
}

impl std::fmt::Debug for Broker {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("Broker")
            .field("network", &self.network)
            .finish_non_exhaustive()
    }
}

impl Broker {
    /// Uses the same holder-to-machine directory as the link host.
    #[must_use]
    pub fn new(
        network: String,
        recorded: BTreeMap<String, MachineIdentity>,
        publisher: Arc<dyn ServicePublisher>,
    ) -> Self {
        Self {
            network,
            recorded,
            registrations: Mutex::new(BTreeMap::new()),
            publisher,
            counters: Arc::new(BrokerCounters::default()),
        }
    }

    /// The counters shared with this broker's echo worker.
    #[must_use]
    pub fn counters(&self) -> Arc<BrokerCounters> {
        Arc::clone(&self.counters)
    }

    /// Decides a control call offline before parsing or changing its state.
    ///
    /// # Errors
    /// Missing call grant, unknown holder, malformed input, absent registration,
    /// poisoned state or a refused publication; each retains its name.
    pub fn serve(
        &self,
        caller: &dyn LysSession,
        subject: &str,
        request: &ServiceRequest<'_>,
        now: u64,
    ) -> Result<Value, BrokerRefusal> {
        if !serves_subject(subject) {
            return Err(BrokerRefusal::Named("not_a_broker_subject"));
        }
        let binding =
            SubjectBinding::new(self.network.clone(), subject.to_owned(), subject.to_owned())
                .map_err(|_| BrokerRefusal::Named("invalid_subject"))?;
        caller.decide(&binding, HotAction::Call, now)?;
        if request.operation.trim().is_empty() || request.operation.len() > 256 {
            return Err(BrokerRefusal::Named("invalid_operation"));
        }
        if request.payload.len() > 16_384 {
            return Err(BrokerRefusal::Named("request_too_large"));
        }
        let device = self
            .recorded
            .get(caller.holder())
            .ok_or(BrokerRefusal::Named("machine_not_recorded"))?;
        match subject {
            "net.broker.register" => self.register(device, request.payload),
            "net.broker.lookup" => self.lookup(request.payload),
            "net.broker.punch" => self.punch(device, caller.holder(), request),
            _ => Err(BrokerRefusal::Named("not_a_broker_subject")),
        }
    }

    fn register(&self, caller: &MachineIdentity, payload: &[u8]) -> Result<Value, BrokerRefusal> {
        let mut registration: Registration = parse(payload)?;
        let device = device_key(&registration.device)?;
        if &device != caller.ed25519_public() {
            return Err(BrokerRefusal::Named("device_mismatch"));
        }
        if registration.addresses.len() > 64
            || registration
                .addresses
                .iter()
                .any(|address| !valid_address(address))
            || registration
                .observed
                .is_some_and(|address| address.port() == 0)
        {
            return Err(BrokerRefusal::Named("invalid_address"));
        }
        registration.device = hex::encode(device);
        self.registrations
            .lock()
            .map_err(|_| BrokerRefusal::Named("registry_poisoned"))?
            .insert(device, registration);
        self.counters.registered.fetch_add(1, Ordering::Relaxed);
        Ok(json!({"ok": true}))
    }

    fn lookup(&self, payload: &[u8]) -> Result<Value, BrokerRefusal> {
        let request: Lookup = parse(payload)?;
        let device = device_key(&request.device)?;
        self.counters.lookups.fetch_add(1, Ordering::Relaxed);
        let registration = {
            let registrations = self
                .registrations
                .lock()
                .map_err(|_| BrokerRefusal::Named("registry_poisoned"))?;
            registrations
                .get(&device)
                .cloned()
                .ok_or(BrokerRefusal::Named("not_registered"))?
        };
        Ok(json!({"addresses": registration.addresses, "observed": registration.observed}))
    }

    fn punch(
        &self,
        caller: &MachineIdentity,
        holder: &str,
        request: &ServiceRequest<'_>,
    ) -> Result<Value, BrokerRefusal> {
        let punch: Punch = parse(request.payload)?;
        let to = device_key(&punch.to)?;
        let nonce = hex::decode(&punch.nonce).map_err(|_| BrokerRefusal::Named("invalid_nonce"))?;
        if nonce.len() != 16 {
            return Err(BrokerRefusal::Named("invalid_nonce"));
        }
        let observed = {
            let registrations = self
                .registrations
                .lock()
                .map_err(|_| BrokerRefusal::Named("registry_poisoned"))?;
            registrations
                .get(caller.ed25519_public())
                .ok_or(BrokerRefusal::Named("not_registered"))?
                .observed
        };
        let payload = json!({"from": hex::encode(caller.ed25519_public()),
            "observed": observed, "nonce": hex::encode(nonce)})
        .to_string();
        // A caller's operation cannot suppress another holder's instruction.
        let operation = blake3::hash(
            &serde_json::to_vec(&(holder, request.operation, to, &payload))
                .map_err(|_| BrokerRefusal::Named("instruction_encoding_failed"))?,
        )
        .to_hex();
        self.publisher.publish(
            &format!("net.punch.{}", hex::encode(to)),
            &service_envelope(b"liminal.net.punch/1", payload.as_bytes()),
            operation.as_str(),
        )?;
        self.counters.punches_told.fetch_add(1, Ordering::Relaxed);
        Ok(json!({"ok": true}))
    }
}

/// The three specified control subjects; arbitrary payload subjects are refused.
#[must_use]
pub fn serves_subject(subject: &str) -> bool {
    matches!(
        subject,
        "net.broker.register" | "net.broker.lookup" | "net.broker.punch"
    )
}

fn parse<T: serde::de::DeserializeOwned>(bytes: &[u8]) -> Result<T, BrokerRefusal> {
    serde_json::from_slice(bytes).map_err(|_| BrokerRefusal::Named("invalid_request"))
}

fn device_key(value: &str) -> Result<[u8; 32], BrokerRefusal> {
    let mut bytes = [0; 32];
    hex::decode_to_slice(value, &mut bytes).map_err(|_| BrokerRefusal::Named("invalid_device"))?;
    MachineIdentity::new(bytes)
        .x25519_public()
        .map_err(|_| BrokerRefusal::Named("invalid_device"))?;
    Ok(bytes)
}

fn valid_address(address: &str) -> bool {
    if address
        .parse::<SocketAddr>()
        .is_ok_and(|address| address.port() != 0)
    {
        return true;
    }
    let Some((host, port)) = address.rsplit_once(':') else {
        return false;
    };
    host.len() <= 253
        && !host.is_empty()
        && host.split('.').all(|label| {
            !label.is_empty()
                && label.len() <= 63
                && !label.starts_with('-')
                && !label.ends_with('-')
                && label
                    .bytes()
                    .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-')
        })
        && port.parse::<u16>().is_ok_and(|port| port != 0)
}

/// Adds broker-specific deterministic configuration refusals.
pub fn validate_configuration(lys: &LysConfig, errors: &mut Vec<String>) {
    let Some(broker) = &lys.broker else {
        return;
    };
    if lys.machine.is_none() {
        errors
            .push("auth.lys.broker: requires auth.lys.machine for recorded caller keys".to_owned());
    }
    if broker.udp_listen.port() == 0 {
        errors.push("auth.lys.broker.udp_listen: port must be nonzero".to_owned());
    }
}

/// One blocking echo worker, stopped by a datagram signal and joined by its owner.
#[derive(Debug)]
pub struct ObservedEcho {
    socket: UdpSocket,
    address: SocketAddr,
    stopping: Arc<AtomicBool>,
    worker: Option<JoinHandle<std::io::Result<()>>>,
}

impl ObservedEcho {
    /// Binds before starting the sole blocking worker; no refresh or idle timer.
    ///
    /// # Errors
    /// The socket could not bind or clone, or the worker could not start.
    pub fn bind(address: SocketAddr, counters: Arc<BrokerCounters>) -> std::io::Result<Self> {
        let socket = UdpSocket::bind(address)?;
        let address = socket.local_addr()?;
        let receive = socket.try_clone()?;
        let stopping = Arc::new(AtomicBool::new(false));
        let stop = Arc::clone(&stopping);
        let worker = std::thread::Builder::new()
            .name("liminal-broker-echo".to_owned())
            .spawn(move || {
                let result = echo_loop(&receive, &counters, &stop);
                if let Err(error) = &result {
                    tracing::error!(%error, "broker echo worker failed");
                }
                result
            })?;
        Ok(Self {
            socket,
            address,
            stopping,
            worker: Some(worker),
        })
    }

    /// The address actually bound, including an allocated ephemeral test port.
    #[must_use]
    pub const fn local_addr(&self) -> SocketAddr {
        self.address
    }

    /// Signals the blocking receive, then joins and reports the worker's result.
    ///
    /// # Errors
    /// The shutdown signal could not send, the worker failed, or it panicked.
    pub fn shutdown(&mut self) -> std::io::Result<()> {
        if self.worker.is_none() {
            return Ok(());
        }
        self.stopping.store(true, Ordering::Release);
        let mut wake = self.address;
        if wake.ip().is_unspecified() {
            wake.set_ip(match wake {
                SocketAddr::V4(_) => IpAddr::V4(Ipv4Addr::LOCALHOST),
                SocketAddr::V6(_) => IpAddr::V6(Ipv6Addr::LOCALHOST),
            });
        }
        self.socket.send_to(&[], wake)?;
        if let Some(worker) = self.worker.take() {
            worker
                .join()
                .map_err(|_| std::io::Error::other("broker echo worker panicked"))??;
        }
        Ok(())
    }
}

impl Drop for ObservedEcho {
    fn drop(&mut self) {
        if let Err(error) = self.shutdown() {
            tracing::error!(%error, "broker echo shutdown failed");
        }
    }
}

fn echo_loop(
    socket: &UdpSocket,
    counters: &BrokerCounters,
    stopping: &AtomicBool,
) -> std::io::Result<()> {
    let mut request = [0; 21];
    loop {
        let (length, source) = match socket.recv_from(&mut request) {
            Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
            result => result?,
        };
        if stopping.load(Ordering::Acquire) {
            return Ok(());
        }
        if length != 20 || &request[..4] != b"LMOB" {
            counters.echo_dropped.fetch_add(1, Ordering::Relaxed);
            continue;
        }
        let mut reply = Vec::with_capacity(39);
        reply.extend_from_slice(b"LMOA");
        reply.extend_from_slice(&request[4..20]);
        match source {
            SocketAddr::V4(address) => {
                reply.push(4);
                reply.extend_from_slice(&address.ip().octets());
            }
            SocketAddr::V6(address) => {
                reply.push(6);
                reply.extend_from_slice(&address.ip().octets());
            }
        }
        reply.extend_from_slice(&source.port().to_be_bytes());
        if socket.send_to(&reply, source)? != reply.len() {
            return Err(std::io::Error::other(
                "broker echo sent an incomplete datagram",
            ));
        }
        counters.echo_answered.fetch_add(1, Ordering::Relaxed);
    }
}