liminal_server/

link_carrier.rs

//! Carriers move sealed link bytes without opening the Noise transport.

use crate::config::LimitsConfig;
use liminal::protocol::{Frame, FrameHeader, ProtocolError, codec};
use std::io::{Read, Write};
use std::net::TcpStream;

pub mod choice;
mod reader;

/// The route a keyed link's bytes take.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Carrier {
    /// The peer's TCP wire listener.
    Wire,
    /// A direct datagram path.
    Direct,
    /// A relay carrying opaque bytes.
    Relay,
}

/// The next sealed message, or the carrier's named end.
#[derive(Debug, PartialEq, Eq)]
pub enum CarrierReceive {
    /// Bytes still sealed by the keyed link.
    Sealed { subject: String, sealed: Vec<u8> },
    /// The importing end acknowledged the message on this link's stream.
    Acknowledged {
        message_id: u64,
        durable_position: Option<u64>,
    },
    /// The importing end refused the message on this link's stream.
    Refused {
        reason_code: u16,
        reason: Option<String>,
    },
    /// The peer or byte stream ended the carrier.
    End { reason: String },
}

/// Why a carrier could not move a frame.
#[derive(Debug, thiserror::Error)]
pub enum CarrierError {
    /// The carrier's owned lifecycle or reader could not continue.
    #[error("wire link lifecycle failed: {0}")]
    Lifecycle(String),
    /// A socket operation failed; credential and message bytes are excluded.
    #[error("wire I/O failed: {0}")]
    Io(#[from] std::io::Error),
    /// The canonical codec refused a frame.
    #[error("wire framing refused: {0}")]
    Protocol(#[from] ProtocolError),
    /// The subject is not a placed name.
    #[error(transparent)]
    Subject(#[from] crate::auth_lys::SubjectBindingError),
    /// A peer's declared frame exceeded the allocation bound.
    #[error("wire frame of {bytes} bytes exceeds bound {bound}")]
    TooLarge { bytes: usize, bound: usize },
    /// The peer answered with a different frame or stream.
    #[error("unexpected frame on wire link")]
    UnexpectedFrame,
    /// Sending after an observed carrier end was refused.
    #[error("wire carrier has ended")]
    Ended,
}

/// A carrier transports sealed bytes and reports its end by name.
pub trait LinkCarrier: Send + std::fmt::Debug {
    /// Which carrier holds this path.
    fn carrier(&self) -> Carrier;
    /// # Errors
    /// A parked reader observed an unsolicited frame or a carrier end.
    fn refuse_unsolicited(&self) -> Result<(), CarrierError> {
        Ok(())
    }
    /// Sends one subject and its sealed bytes.
    ///
    /// # Errors
    /// The frame was invalid or the carrier could not send it.
    fn send(&mut self, subject: &str, sealed: &[u8]) -> Result<(), CarrierError>;
    /// Receives the next sealed bytes, or the carrier's named end.
    ///
    /// # Errors
    /// An I/O error, malformed frame, or unexpected frame/stream.
    fn receive(&mut self) -> Result<CarrierReceive, CarrierError>;
}

/// One application stream on the peer's TCP wire connection.
#[derive(Debug)]
pub struct WireCarrier {
    socket: TcpStream,
    stream_id: u32,
    ended: Option<String>,
    life: Option<crate::link::LinkLife>,
    reader: Option<reader::Reader>,
}

impl WireCarrier {
    pub(crate) fn refuse_unsolicited(&self) -> Result<(), CarrierError> {
        if let Some(reader) = &self.reader
            && let Some(answer) = reader.try_receive()?
        {
            let error = match answer {
                Ok(Some(_)) => CarrierError::UnexpectedFrame,
                Ok(None) => CarrierError::Ended,
                Err(error) => error,
            };
            self.failed(&error);
            return Err(error);
        }
        Ok(())
    }
    pub(crate) fn install_life(&mut self, life: crate::link::LinkLife) -> Result<(), CarrierError> {
        if self.life.is_some() {
            return Err(CarrierError::Lifecycle(
                "lifecycle already installed".to_owned(),
            ));
        }
        self.life = Some(life.clone());
        match reader::Reader::start(&self.socket, self.stream_id, life) {
            Ok(reader) => self.reader = Some(reader),
            Err(error) => {
                self.end_life(crate::link::LinkEnd::Refused(error.to_string()));
                return Err(error);
            }
        }
        Ok(())
    }

    pub(crate) const fn life(&self) -> Option<&crate::link::LinkLife> {
        self.life.as_ref()
    }

    fn end_life(&self, reason: crate::link::LinkEnd) {
        if let Some(life) = &self.life
            && let Err(error) = life.end(reason)
        {
            tracing::error!(%error, "wire link end observation failed");
        }
    }

    fn failed(&self, error: &CarrierError) {
        self.end_life(
            if matches!(error, CarrierError::Io(_) | CarrierError::Ended) {
                crate::link::LinkEnd::PeerGone
            } else {
                crate::link::LinkEnd::Refused(error.to_string())
            },
        );
    }
    /// Duplicates the carrier socket solely to interrupt its blocking receive at stop.
    ///
    /// # Errors
    /// The socket cannot be duplicated.
    pub fn interrupt_handle(&self) -> Result<TcpStream, CarrierError> {
        Ok(self.socket.try_clone()?)
    }
    /// Takes ownership of a connection whose link handshake has finished.
    ///
    /// # Errors
    /// Stream zero is reserved for connection control.
    pub fn new(socket: TcpStream, stream_id: u32) -> Result<Self, CarrierError> {
        liminal::protocol::validate_stream(liminal::protocol::FrameType::LinkMessage, stream_id)?;
        Ok(Self {
            socket,
            stream_id,
            ended: None,
            life: None,
            reader: None,
        })
    }
}

impl LinkCarrier for WireCarrier {
    fn refuse_unsolicited(&self) -> Result<(), CarrierError> {
        Self::refuse_unsolicited(self)
    }
    fn carrier(&self) -> Carrier {
        Carrier::Wire
    }

    fn send(&mut self, subject: &str, sealed: &[u8]) -> Result<(), CarrierError> {
        if self.ended.is_some() {
            return Err(CarrierError::Ended);
        }
        if let Some(life) = &self.life
            && life.reason().map_err(CarrierError::Lifecycle)?.is_some()
        {
            return Err(CarrierError::Ended);
        }
        crate::auth_lys::SubjectBinding::new(
            "wire".to_owned(),
            subject.to_owned(),
            subject.to_owned(),
        )?;
        let result = write_frame(
            &mut self.socket,
            &Frame::LinkMessage {
                flags: 0,
                stream_id: self.stream_id,
                subject: subject.to_owned(),
                sealed: sealed.to_vec(),
            },
        );
        if let Err(error) = &result {
            self.failed(error);
        }
        result
    }

    fn receive(&mut self) -> Result<CarrierReceive, CarrierError> {
        if let Some(reason) = &self.ended {
            return Ok(CarrierReceive::End {
                reason: reason.clone(),
            });
        }
        let answer = if let Some(reader) = &self.reader {
            reader.receive()
        } else {
            read_frame(&mut self.socket)
        };
        let frame = answer.inspect_err(|error| self.failed(error))?;
        let reason = match frame {
            Some(Frame::PublishAck {
                stream_id,
                message_id,
                durable_position,
                ..
            }) if stream_id == self.stream_id => {
                return Ok(CarrierReceive::Acknowledged {
                    message_id,
                    durable_position,
                });
            }
            Some(Frame::PublishError {
                stream_id,
                reason_code,
                message,
                ..
            }) if stream_id == self.stream_id => {
                return Ok(CarrierReceive::Refused {
                    reason_code,
                    reason: message,
                });
            }
            Some(Frame::LinkMessage {
                stream_id,
                subject,
                sealed,
                ..
            }) if stream_id == self.stream_id => {
                crate::auth_lys::SubjectBinding::new(
                    "wire".to_owned(),
                    subject.clone(),
                    subject.clone(),
                )?;
                return Ok(CarrierReceive::Sealed { subject, sealed });
            }
            Some(Frame::DisconnectReason { reason, .. }) => reason.name().to_owned(),
            Some(Frame::Disconnect { .. }) => "peer_disconnect".to_owned(),
            None => "wire_eof".to_owned(),
            Some(_) => {
                self.failed(&CarrierError::UnexpectedFrame);
                return Err(CarrierError::UnexpectedFrame);
            }
        };
        self.end_life(
            if reason == "grant_revoked" || reason == "revocation_log_reset" {
                crate::link::LinkEnd::Revoked
            } else {
                crate::link::LinkEnd::PeerGone
            },
        );
        self.ended = Some(reason.clone());
        Ok(CarrierReceive::End { reason })
    }
}

impl Drop for WireCarrier {
    fn drop(&mut self) {
        self.end_life(crate::link::LinkEnd::Shutdown);
        if let Some(reader) = self.reader.take() {
            reader.close(&self.socket);
        }
    }
}

/// An ordinary wire connection to the relay on its admitted relay stream.
#[derive(Debug)]
pub struct RelayCarrier {
    wire: WireCarrier,
    relay: crate::link_handshake::MachineIdentity,
}

impl RelayCarrier {
    /// Takes a connection whose relay stream has been granted and marked.
    ///
    /// # Errors
    /// Stream zero is reserved for connection control.
    pub fn new(
        socket: TcpStream,
        stream_id: u32,
        relay: crate::link_handshake::MachineIdentity,
    ) -> Result<Self, CarrierError> {
        Ok(Self {
            wire: WireCarrier::new(socket, stream_id)?,
            relay,
        })
    }

    /// Interrupts a blocking receive through the same owned socket.
    ///
    /// # Errors
    /// The socket cannot be duplicated.
    pub fn interrupt_handle(&self) -> Result<TcpStream, CarrierError> {
        self.wire.interrupt_handle()
    }

    /// Answers an imported frame; the relay carries the full receipt back.
    ///
    /// # Errors
    /// The carrier has ended or the ordinary connection cannot send the ack.
    pub fn acknowledge(
        &mut self,
        message_id: u64,
        durable_position: Option<u64>,
    ) -> Result<(), CarrierError> {
        self.reply(&Frame::PublishAck {
            flags: 0,
            stream_id: self.wire.stream_id,
            message_id,
            durable_position,
        })
    }

    /// Answers a named endpoint refusal on the relay stream.
    ///
    /// # Errors
    /// The carrier has ended or the ordinary connection cannot send the refusal.
    pub fn refuse(&mut self, reason_code: u16, reason: &str) -> Result<(), CarrierError> {
        self.reply(&Frame::PublishError {
            flags: 0,
            stream_id: self.wire.stream_id,
            reason_code,
            message: Some(reason.to_owned()),
        })
    }

    /// The topology record for the endpoint's observed relayed link.
    #[must_use]
    pub fn topology(
        &self,
        local: crate::link_handshake::MachineIdentity,
        peer: crate::link_handshake::MachineIdentity,
        counters: crate::console::records::LinkCounters,
    ) -> crate::console::records::Link {
        crate::console::records::Link::federated(
            &hex::encode(local.ed25519_public()),
            &hex::encode(peer.ed25519_public()),
            Some(crate::console::records::Carrier::Relay(hex::encode(
                self.relay.ed25519_public(),
            ))),
            crate::console::records::LinkState::Up,
            None,
            counters,
        )
    }

    fn reply(&mut self, frame: &Frame) -> Result<(), CarrierError> {
        if self.wire.ended.is_some() {
            return Err(CarrierError::Ended);
        }
        write_frame(&mut self.wire.socket, frame)
    }
}

impl LinkCarrier for RelayCarrier {
    fn refuse_unsolicited(&self) -> Result<(), CarrierError> {
        self.wire.refuse_unsolicited()
    }
    fn carrier(&self) -> Carrier {
        Carrier::Relay
    }
    fn send(&mut self, subject: &str, sealed: &[u8]) -> Result<(), CarrierError> {
        self.wire.send(subject, sealed)
    }
    fn receive(&mut self) -> Result<CarrierReceive, CarrierError> {
        self.wire.receive()
    }
}

/// The same bound is enforced before encoding and before allocating a peer's body.
pub(crate) const WIRE_FRAME_BOUND: usize = LimitsConfig::DEFAULT_MAX_CONNECTION_INBOX_BYTES;

pub(crate) fn write_frame(socket: &mut TcpStream, frame: &Frame) -> Result<(), CarrierError> {
    let bytes = codec::encoded_len(frame)?;
    if bytes > WIRE_FRAME_BOUND {
        return Err(CarrierError::TooLarge {
            bytes,
            bound: WIRE_FRAME_BOUND,
        });
    }
    let mut encoded = vec![0; bytes];
    let length = codec::encode(frame, &mut encoded)?;
    socket.write_all(&encoded[..length])?;
    Ok(())
}

pub(crate) fn read_frame(socket: &mut TcpStream) -> Result<Option<Frame>, CarrierError> {
    let mut header = [0; FrameHeader::WIRE_LEN];
    // Only EOF before a header is a clean end; an incomplete header/body is an error.
    loop {
        match socket.read(&mut header[..1]) {
            Ok(0) => return Ok(None),
            Ok(_) => break,
            Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
            Err(error) => return Err(error.into()),
        }
    }
    socket.read_exact(&mut header[1..])?;
    let payload = u32::from_be_bytes([header[6], header[7], header[8], header[9]]);
    let bytes = usize::try_from(payload)
        .ok()
        .and_then(|length| length.checked_add(header.len()))
        .ok_or(CarrierError::TooLarge {
            bytes: usize::MAX,
            bound: WIRE_FRAME_BOUND,
        })?;
    if bytes > WIRE_FRAME_BOUND {
        return Err(CarrierError::TooLarge {
            bytes,
            bound: WIRE_FRAME_BOUND,
        });
    }
    let mut encoded = vec![0; bytes];
    encoded[..header.len()].copy_from_slice(&header);
    socket.read_exact(&mut encoded[header.len()..])?;
    let (frame, _) = codec::decode(&encoded)?;
    Ok(Some(frame))
}