liminal-server 0.25.4

//! 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 relay;
pub use relay::RelayCarrier;
#[cfg(test)]
#[path = "link_carrier/ended_tests.rs"]
mod ended_tests;
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(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 },
}

impl std::fmt::Debug for CarrierReceive {
    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Sealed { subject, sealed } => out
                .debug_struct("Sealed")
                .field("subject", subject)
                .field("bytes", &sealed.len())
                .finish(),
            Self::Acknowledged {
                message_id,
                durable_position,
            } => out
                .debug_struct("Acknowledged")
                .field("message_id", message_id)
                .field("durable_position", durable_position)
                .finish(),
            Self::Refused {
                reason_code,
                reason,
            } => out
                .debug_struct("Refused")
                .field("reason_code", reason_code)
                .field("reason_bytes", &reason.as_ref().map(String::len))
                .finish(),
            Self::End { reason } => out
                .debug_struct("End")
                .field("reason_bytes", &reason.len())
                .finish(),
        }
    }
}

/// 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>;
    /// The end this carrier observed, by name, once it has one: the peer's
    /// close reason, `wire_eof`, or the failure that ended it. `None` while
    /// the carrier is live.
    fn ended(&self) -> Option<String> {
        None
    }
}

/// One application stream on the peer's TCP wire connection.
#[derive(Debug)]
pub struct WireCarrier {
    socket: TcpStream,
    stream_id: u32,
    ended: Option<String>,
    /// The first failure or end observed, by name: by a send or receive on
    /// the owning thread, or by the reader thread as it reads the end.
    failure: std::sync::Arc<std::sync::OnceLock<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,
            std::sync::Arc::clone(&self.failure),
        ) {
            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.failure.get_or_init(|| error.to_string());
        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,
            failure: std::sync::Arc::new(std::sync::OnceLock::new()),
            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 ended(&self) -> Option<String> {
        self.ended
            .clone()
            .or_else(|| self.failure.get().cloned())
            .or_else(|| {
                self.life
                    .as_ref()
                    .and_then(|life| life.reason().ok().flatten())
            })
    }

    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);
        }
    }
}

/// 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))
}