liminal-server 0.25.4

//! Only the endpoints key and open a relayed Noise link.

use crate::link_carrier::{CarrierError, CarrierReceive, LinkCarrier, RelayCarrier};
use crate::link_handshake::{
    LinkHandshake, LinkHandshakeError, LinkTransport, LocalMachine, MachineIdentity, NOISE_OVERHEAD,
};

const HANDSHAKE_SUBJECT: &str = "net.link.handshake";

/// The proved endpoint transport and its opaque relay path.
#[derive(Debug)]
pub struct RelayedLink {
    /// Noise keys known only to the two linked machines.
    pub transport: LinkTransport,
    /// The admitted relay connection and marked application stream.
    pub carrier: RelayCarrier,
}

/// A relayed open retains the endpoint or relay's own refusal.
#[derive(Debug, thiserror::Error)]
pub enum RelayOpenError {
    /// The ordinary wire connection failed or answered an unexpected frame.
    #[error(transparent)]
    Carrier(#[from] CarrierError),
    /// The endpoint's Noise peer proof failed.
    #[error(transparent)]
    Handshake(#[from] LinkHandshakeError),
    /// The relay or the far endpoint refused or ended the open.
    #[error("{0}")]
    Refused(String),
}

/// Sends the opening KK message over the same relay stream as later frames.
///
/// # Errors
/// Wire failure, relay refusal, unexpected handshake traffic or failed peer proof.
pub fn open(
    mut carrier: RelayCarrier,
    local: &LocalMachine,
    peer: MachineIdentity,
) -> Result<RelayedLink, RelayOpenError> {
    let mut handshake = LinkHandshake::initiator(local, peer)?;
    send_handshake(&mut carrier, &mut handshake)?;
    let mut acknowledged = false;
    let mut received = false;
    while !acknowledged || !received {
        match next(&mut carrier)? {
            CarrierReceive::Acknowledged { .. } if !acknowledged => acknowledged = true,
            CarrierReceive::Sealed { subject, sealed }
                if !received && subject == HANDSHAKE_SUBJECT =>
            {
                read_handshake(&mut carrier, &mut handshake, &sealed)?;
                received = true;
            }
            _ => return Err(CarrierError::UnexpectedFrame.into()),
        }
    }
    Ok(RelayedLink {
        transport: handshake.finish()?,
        carrier,
    })
}

/// Answers the opening KK message without granting the relay any key material.
///
/// # Errors
/// Wire failure, relay refusal, unexpected handshake traffic or failed peer proof.
pub fn accept(
    mut carrier: RelayCarrier,
    local: &LocalMachine,
    peer: MachineIdentity,
) -> Result<RelayedLink, RelayOpenError> {
    let mut handshake = LinkHandshake::responder(local, peer)?;
    match next(&mut carrier)? {
        CarrierReceive::Sealed { subject, sealed } if subject == HANDSHAKE_SUBJECT => {
            read_handshake(&mut carrier, &mut handshake, &sealed)?;
        }
        _ => return Err(CarrierError::UnexpectedFrame.into()),
    }
    send_handshake(&mut carrier, &mut handshake)?;
    if !matches!(next(&mut carrier)?, CarrierReceive::Acknowledged { .. }) {
        return Err(CarrierError::UnexpectedFrame.into());
    }
    Ok(RelayedLink {
        transport: handshake.finish()?,
        carrier,
    })
}

fn send_handshake(
    carrier: &mut RelayCarrier,
    handshake: &mut LinkHandshake,
) -> Result<(), RelayOpenError> {
    let mut bytes = [0; NOISE_OVERHEAD];
    let length = handshake.write(&mut bytes)?;
    carrier.send(HANDSHAKE_SUBJECT, &bytes[..length])?;
    Ok(())
}

fn read_handshake(
    carrier: &mut RelayCarrier,
    handshake: &mut LinkHandshake,
    bytes: &[u8],
) -> Result<(), RelayOpenError> {
    if let Err(error) = handshake.read(bytes) {
        carrier.refuse(0x0302, "relay_handshake_refused")?;
        return Err(error.into());
    }
    carrier.acknowledge(0, None)?;
    Ok(())
}

fn next(carrier: &mut RelayCarrier) -> Result<CarrierReceive, RelayOpenError> {
    match carrier.receive()? {
        CarrierReceive::Refused {
            reason_code,
            reason,
        } => {
            Err(RelayOpenError::Refused(reason.unwrap_or_else(|| {
                format!("relay_publish_error_{reason_code}")
            })))
        }
        CarrierReceive::End { reason } => Err(RelayOpenError::Refused(reason)),
        received => Ok(received),
    }
}

#[cfg(feature = "test-support")]
impl crate::server::connection::ConnectionSupervisor {
    /// Boots a relay fixture with its configured machine and service bindings.
    ///
    /// # Errors
    /// Configuration, machine admission, relay binding or scheduler startup failed.
    pub fn with_relay_for_tests(
        config: &crate::config::ServerConfig,
        services: std::sync::Arc<crate::server::connection::LiminalConnectionServices>,
        handle: tokio::runtime::Handle,
    ) -> Result<Self, crate::ServerError> {
        let refused = |message: &str| crate::ServerError::ConfigValidation {
            message: message.to_owned(),
        };
        let auth = config.auth.as_ref().ok_or_else(|| refused("auth absent"))?;
        let lys = auth
            .lys
            .as_ref()
            .ok_or_else(|| refused("auth.lys absent"))?;
        let mut install = crate::server::connection::configured_lys(auth, Some(handle.clone()))?
            .ok_or_else(|| refused("Lys install absent"))?;
        if install.link_host.is_none() {
            return Err(refused("relay fixture has no installed machine identity"));
        }
        let bindings = crate::services::production::ServiceBindings::configure(
            lys,
            std::sync::Arc::clone(&services),
            handle,
            1,
        )
        .map_err(|error| crate::ServerError::ConfigValidation {
            message: error.to_string(),
        })?
        .ok_or_else(|| refused("relay bindings absent"))?;
        if bindings.relay().is_none() {
            return Err(refused("relay not configured"));
        }
        install.services = Some(std::sync::Arc::new(bindings));
        let observations = std::sync::Arc::clone(&services);
        let supervisor = Self::with_lys_and_fatal_shutdown(
            services,
            None,
            None,
            Some(install),
            config.limits,
            crate::server::shutdown::ShutdownHandle::new(),
        )?;
        supervisor.install_link_observations_for_tests(&observations)?;
        Ok(supervisor)
    }
}