liminal-server 0.24.5

//! The production bindings of the service seams (ACCESS-003 R1), built once
//! from `[auth.lys.services]` and `[auth.lys.broker]`.
//!
//! - [`SeatStarter`](super::SeatStarter): [`LysSeatStarter`], a seat started
//!   by Lys's start-command route on the configured machine.
//! - [`KeySender`](super::KeySender): [`ChannelKeySender`], the keys
//!   published on `sessions.<id>.keys` through the server's own publish path.
//! - [`BudgetSource`](super::BudgetSource): [`LysBudgets`], budgets and stops
//!   read from Lys, carried whole.
//! - [`ServiceAction`](super::ServiceAction): [`PublishingActions`], alarms
//!   and launches published on their subjects, context readings reported to
//!   Lys through the [`ContextReporter`] the context watch shares.
//!
//! Every Lys request is made as this liminal, a machine Lys knows: the
//! bearer is the current machine pass held by the boot pass source,
//! verified offline at start exactly as any presented pass is. Without
//! `[auth.lys.machine]` the services cannot ask Lys as anyone, and start is
//! refused by that name, never served with a guessed identity.

use super::actions::PublishingActions;
use super::broker::{Broker, BrokerRefusal, ObservedEcho};
use super::budgets::LysBudgets;
use super::context_watch::ContextReporter;
use super::keys::{ChannelKeySender, ConnectionPublisher, ServicePublisher};
use super::lys_http::{HttpLys, LysAnswer, LysHttp};
use super::seats::LysSeatStarter;
use super::{
    BudgetPolicy, BudgetSource, KeySender, SeatStarter, ServiceAction, ServiceEffectError,
    ServiceKind, ServiceRequest, Services,
};
use crate::auth_lys::LysPassConsumer;
use crate::config::types::{LysConfig, MachineConfig};
use crate::link_pass::MachinePass;
use crate::server::connection::LiminalConnectionServices;
use std::net::SocketAddr;
use std::sync::Arc;
mod debug;

/// Administrative effect seams and an optional address broker, bound once.
pub struct ServiceBindings {
    administrative: Option<AdministrativeBindings>,
    context: Arc<ContextReporter>,
    watch_context: bool,
    broker: Option<Broker>,
    echo: Option<ObservedEcho>,
    relay: Option<Arc<super::relay::Relay>>,
    relay_topology_local: Option<String>,
}

#[derive(Debug)]
struct AdministrativeBindings {
    seats: LysSeatStarter,
    keys: ChannelKeySender,
    budgets: LysBudgets,
    actions: PublishingActions,
}

impl std::fmt::Debug for ServiceBindings {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("ServiceBindings")
            .field("administrative", &self.administrative)
            .field("watch_context", &self.watch_context)
            .field("broker", &self.broker)
            .field("echo", &self.echo)
            .field("relay", &self.relay)
            .finish_non_exhaustive()
    }
}

impl ServiceBindings {
    /// Binds every seam: Lys asked through `http` with `bearer` (this
    /// machine's own pass), seats started on `machine`, messages published
    /// through `publisher`.
    #[must_use]
    pub fn new(
        http: &Arc<dyn LysHttp>,
        bearer: &str,
        machine: String,
        publisher: &Arc<dyn ServicePublisher>,
    ) -> Self {
        let context = Arc::new(ContextReporter::new(Arc::clone(http), bearer.to_owned()));
        Self {
            administrative: Some(AdministrativeBindings {
                seats: LysSeatStarter::new(Arc::clone(http), bearer.to_owned(), machine),
                keys: ChannelKeySender::new(Arc::clone(publisher)),
                budgets: LysBudgets::new(Arc::clone(http), bearer.to_owned()),
                actions: PublishingActions::new(Arc::clone(publisher), Arc::clone(&context)),
            }),
            context,
            watch_context: false,
            broker: None,
            echo: None,
            relay: None,
            relay_topology_local: None,
        }
    }

    /// Whether the operator asked for the context watch to run.
    #[must_use]
    pub const fn watches_context(&self) -> bool {
        self.watch_context
    }

    /// The seams, lent for one [`serve`](super::serve).
    #[must_use]
    pub fn services(&self) -> Services<'_> {
        Services {
            seats: self,
            keys: self,
            budgets: self,
            actions: self,
        }
    }

    /// The reporter the context watch shares with the `context-watch` call.
    #[must_use]
    pub fn context_reporter(&self) -> Arc<ContextReporter> {
        Arc::clone(&self.context)
    }

    /// The configured address broker, independent of administrative services.
    #[must_use]
    pub const fn broker(&self) -> Option<&Broker> {
        self.broker.as_ref()
    }

    /// The address actually bound by the configured broker's UDP echo.
    #[must_use]
    pub fn echo_address(&self) -> Option<SocketAddr> {
        self.echo.as_ref().map(ObservedEcho::local_addr)
    }

    /// The configured blind relay, independent of administrative services.
    #[must_use]
    pub const fn relay(&self) -> Option<&Arc<super::relay::Relay>> {
        self.relay.as_ref()
    }

    /// The console's canonical local endpoint for relay observations.
    #[must_use]
    pub fn relay_topology_local(&self) -> Option<&str> {
        self.relay_topology_local.as_deref()
    }

    /// Serves a broker request on the connection's admitted session.
    ///
    /// # Errors
    /// The broker or admitted session is absent, or the broker refuses the call.
    pub fn serve_broker(
        &self,
        session: Option<&dyn crate::auth_lys::session::LysSession>,
        subject: &str,
        request: &ServiceRequest<'_>,
        now: u64,
    ) -> Result<serde_json::Value, BrokerRefusal> {
        let broker = self
            .broker
            .as_ref()
            .ok_or(BrokerRefusal::Named("broker_not_configured"))?;
        let session = session.ok_or(BrokerRefusal::Named("pass_not_admitted"))?;
        broker.serve(session, subject, request, now)
    }

    /// Builds configured services when the boot machine pass is present.
    /// `Ok(None)` when no services are selected or a dial pass was refused.
    /// Lys is asked on `handle`'s runtime; messages are published through
    /// `services` as `liminal.<host>`.
    ///
    /// # Errors
    /// `[auth.lys.machine]` is absent (the services would have no identity to
    /// ask Lys as), this machine's pass or the issuer's key set cannot be
    /// read or does not verify, or the issuer is not an HTTP origin.
    pub fn configure(
        lys: &LysConfig,
        services: Arc<LiminalConnectionServices>,
        handle: tokio::runtime::Handle,
        now: u64,
    ) -> Result<Option<Self>, ServiceEffectError> {
        if lys.services.is_none()
            && lys.broker.is_none()
            && !lys.relay.as_ref().is_some_and(|relay| relay.enabled)
        {
            return Ok(None);
        }
        let machine = lys
            .machine
            .as_ref()
            .ok_or_else(|| ServiceEffectError("auth.lys.machine is absent".into()))?;
        let Some(pass) = services_pass(machine)? else {
            return Ok(None);
        };
        let bearer = machine_bearer(lys, now)?;
        let http: Arc<dyn LysHttp> = Arc::new(MachineHttp {
            http: HttpLys::new(lys.api_url.as_deref().unwrap_or(&lys.issuer), handle)?,
            pass,
        });
        let host = lys
            .services
            .as_ref()
            .map_or("broker", |section| section.host.as_str());
        let topology_local = services.console().device().to_owned();
        let publisher: Arc<dyn ServicePublisher> =
            Arc::new(ConnectionPublisher::new(services, host.to_owned()));
        let mut bindings = if let Some(section) = &lys.services {
            let mut bindings = Self::new(&http, &bearer, section.machine.clone(), &publisher);
            bindings.watch_context = section.context_watch;
            bindings
        } else {
            Self {
                administrative: None,
                context: Arc::new(ContextReporter::new(Arc::clone(&http), bearer)),
                watch_context: false,
                broker: None,
                echo: None,
                relay: None,
                relay_topology_local: None,
            }
        };
        if let Some(section) = &lys.broker {
            let machine = lys.machine.as_ref().ok_or_else(|| {
                ServiceEffectError("auth.lys.broker requires auth.lys.machine".to_owned())
            })?;
            let recorded = machine
                .machines
                .iter()
                .map(|(holder, key)| {
                    let mut bytes = [0; 32];
                    hex::decode_to_slice(key, &mut bytes).map_err(|_| {
                        ServiceEffectError(format!(
                            "auth.lys.machine.machines: invalid key for holder {holder}"
                        ))
                    })?;
                    let identity = crate::link_handshake::MachineIdentity::new(bytes);
                    identity.x25519_public().map_err(|error| {
                        ServiceEffectError(format!(
                            "auth.lys.machine.machines: holder {holder}: {error}"
                        ))
                    })?;
                    Ok((holder.clone(), identity))
                })
                .collect::<Result<_, ServiceEffectError>>()?;
            let broker = Broker::new(lys.network_id.clone(), recorded, publisher);
            bindings.echo = Some(
                ObservedEcho::bind(section.udp_listen, broker.counters()).map_err(|error| {
                    ServiceEffectError(format!("auth.lys.broker.udp_listen: {error}"))
                })?,
            );
            bindings.broker = Some(broker);
        }
        if lys.relay.as_ref().is_some_and(|relay| relay.enabled) {
            let machine = lys.machine.as_ref().ok_or_else(|| {
                ServiceEffectError("auth.lys.relay requires auth.lys.machine".to_owned())
            })?;
            let recorded = machine
                .machines
                .iter()
                .map(|(holder, key)| {
                    let mut bytes = [0; 32];
                    hex::decode_to_slice(key, &mut bytes).map_err(|_| {
                        ServiceEffectError(format!("invalid recorded machine for holder {holder}"))
                    })?;
                    let identity = crate::link_handshake::MachineIdentity::new(bytes);
                    identity
                        .x25519_public()
                        .map_err(|error| ServiceEffectError(error.to_string()))?;
                    Ok((holder.clone(), identity))
                })
                .collect::<Result<_, ServiceEffectError>>()?;
            bindings.relay = Some(super::relay::Relay::new(lys.network_id.clone(), recorded));
            bindings.relay_topology_local = Some(topology_local);
        }
        Ok(Some(bindings))
    }

    fn administrative(&self) -> Result<&AdministrativeBindings, ServiceEffectError> {
        self.administrative
            .as_ref()
            .ok_or_else(|| ServiceEffectError("auth.lys.services is not configured".to_owned()))
    }
}

impl SeatStarter for ServiceBindings {
    fn start(&self, seat: &str, operation: &str) -> Result<(), ServiceEffectError> {
        self.administrative()?.seats.start(seat, operation)
    }
}

impl KeySender for ServiceBindings {
    fn send(&self, session: &str, keys: &[u8], operation: &str) -> Result<(), ServiceEffectError> {
        self.administrative()?.keys.send(session, keys, operation)
    }
}

impl BudgetSource for ServiceBindings {
    fn read(&self, name: &str) -> Result<Option<BudgetPolicy>, ServiceEffectError> {
        self.administrative()?.budgets.read(name)
    }
}

impl ServiceAction for ServiceBindings {
    fn act(
        &self,
        service: ServiceKind,
        payload: &[u8],
        operation: &str,
    ) -> Result<(), ServiceEffectError> {
        self.administrative()?
            .actions
            .act(service, payload, operation)
    }
}

/// The pass the services present, or `None` when a dial machine holds none
/// yet: its ask was refused at start (ACCESS-009 R1, the link asks again at
/// its next open) or no source is configured. Services bind at start only.
///
/// # Errors
/// The pass holder is poisoned, or the pass cannot be read for a reason
/// other than an absent source.
fn services_pass(machine: &MachineConfig) -> Result<Option<MachinePass>, ServiceEffectError> {
    match MachinePass::configured(machine) {
        Ok(pass) => {
            let issued = pass
                .issued()
                .map_err(|error| ServiceEffectError(error.to_string()))?;
            Ok((issued || machine.agent_id.is_empty()).then_some(pass))
        }
        Err(error) if !machine.agent_id.is_empty() && error.source_absent() => Ok(None),
        Err(error) => Err(ServiceEffectError(error.to_string())),
    }
}

/// This machine's own Lys pass as the bearer the services ask Lys with.
///
/// It is verified offline for audience `liminal` at `now` against the
/// issuer's key set, so an expired or foreign pass refuses start by its
/// cause.
///
/// # Errors
/// No `[auth.lys.machine]`, or the pass or key set is unreadable or refused.
pub fn machine_bearer(lys: &LysConfig, now: u64) -> Result<String, ServiceEffectError> {
    let machine = lys.machine.as_ref().ok_or_else(|| {
        ServiceEffectError(
            "auth.lys.services or auth.lys.broker needs auth.lys.machine: services use this machine's pass"
                .to_owned(),
        )
    })?;
    let keys_json = std::fs::read_to_string(&lys.keys_path).map_err(|error| {
        ServiceEffectError(format!(
            "auth.lys.keys_path {}: {error}",
            lys.keys_path.display()
        ))
    })?;
    let adapter = crate::auth_lys::lys_pass_adapter::LysPassAdapter::from_keys_json(
        &keys_json,
        lys.issuer.clone(),
    )
    .map_err(|error| ServiceEffectError(format!("auth.lys key set refused: {error}")))?;
    let pass = crate::link_pass::MachinePass::configured(machine)
        .map_err(|error| ServiceEffectError(error.to_string()))?;
    let pass_bytes = pass
        .current()
        .map_err(|error| ServiceEffectError(error.to_string()))?;
    adapter
        .verify_offline(
            pass_bytes.as_bytes(),
            crate::auth_lys::LIMINAL_PASS_AUDIENCE,
            now,
        )
        .map_err(|error| ServiceEffectError(format!("auth.lys.machine pass refused: {error}")))?;
    Ok(pass_bytes.trim().to_owned())
}

// Keep the holder so renewal reaches services that are already bound.
struct MachineHttp {
    http: HttpLys,
    pass: crate::link_pass::MachinePass,
}
impl LysHttp for MachineHttp {
    fn get(&self, _: &str, path: &str) -> Result<LysAnswer, ServiceEffectError> {
        let token = self
            .pass
            .current()
            .map_err(|error| ServiceEffectError(error.to_string()))?;
        self.http.get(&token, path)
    }
    fn post(
        &self,
        _: &str,
        path: &str,
        body: &serde_json::Value,
    ) -> Result<LysAnswer, ServiceEffectError> {
        let token = self
            .pass
            .current()
            .map_err(|error| ServiceEffectError(error.to_string()))?;
        self.http.post(&token, path, body)
    }
}

#[cfg(test)]
mod tests;