liminal_server/
link_carrier.rs

1//! Carriers move sealed link bytes without opening the Noise transport.
2
3use crate::config::LimitsConfig;
4use liminal::protocol::{Frame, FrameHeader, ProtocolError, codec};
5use std::io::{Read, Write};
6use std::net::TcpStream;
7
8pub mod choice;
9mod relay;
10pub use relay::RelayCarrier;
11#[cfg(test)]
12#[path = "link_carrier/ended_tests.rs"]
13mod ended_tests;
14mod reader;
15
16/// The route a keyed link's bytes take.
17#[derive(Clone, Copy, Debug, PartialEq, Eq)]
18pub enum Carrier {
19    /// The peer's TCP wire listener.
20    Wire,
21    /// A direct datagram path.
22    Direct,
23    /// A relay carrying opaque bytes.
24    Relay,
25}
26
27/// The next sealed message, or the carrier's named end.
28#[derive(PartialEq, Eq)]
29pub enum CarrierReceive {
30    /// Bytes still sealed by the keyed link.
31    Sealed { subject: String, sealed: Vec<u8> },
32    /// The importing end acknowledged the message on this link's stream.
33    Acknowledged {
34        message_id: u64,
35        durable_position: Option<u64>,
36    },
37    /// The importing end refused the message on this link's stream.
38    Refused {
39        reason_code: u16,
40        reason: Option<String>,
41    },
42    /// The peer or byte stream ended the carrier.
43    End { reason: String },
44}
45
46impl std::fmt::Debug for CarrierReceive {
47    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
48        match self {
49            Self::Sealed { subject, sealed } => out
50                .debug_struct("Sealed")
51                .field("subject", subject)
52                .field("bytes", &sealed.len())
53                .finish(),
54            Self::Acknowledged {
55                message_id,
56                durable_position,
57            } => out
58                .debug_struct("Acknowledged")
59                .field("message_id", message_id)
60                .field("durable_position", durable_position)
61                .finish(),
62            Self::Refused {
63                reason_code,
64                reason,
65            } => out
66                .debug_struct("Refused")
67                .field("reason_code", reason_code)
68                .field("reason_bytes", &reason.as_ref().map(String::len))
69                .finish(),
70            Self::End { reason } => out
71                .debug_struct("End")
72                .field("reason_bytes", &reason.len())
73                .finish(),
74        }
75    }
76}
77
78/// Why a carrier could not move a frame.
79#[derive(Debug, thiserror::Error)]
80pub enum CarrierError {
81    /// The carrier's owned lifecycle or reader could not continue.
82    #[error("wire link lifecycle failed: {0}")]
83    Lifecycle(String),
84    /// A socket operation failed; credential and message bytes are excluded.
85    #[error("wire I/O failed: {0}")]
86    Io(#[from] std::io::Error),
87    /// The canonical codec refused a frame.
88    #[error("wire framing refused: {0}")]
89    Protocol(#[from] ProtocolError),
90    /// The subject is not a placed name.
91    #[error(transparent)]
92    Subject(#[from] crate::auth_lys::SubjectBindingError),
93    /// A peer's declared frame exceeded the allocation bound.
94    #[error("wire frame of {bytes} bytes exceeds bound {bound}")]
95    TooLarge { bytes: usize, bound: usize },
96    /// The peer answered with a different frame or stream.
97    #[error("unexpected frame on wire link")]
98    UnexpectedFrame,
99    /// Sending after an observed carrier end was refused.
100    #[error("wire carrier has ended")]
101    Ended,
102}
103
104/// A carrier transports sealed bytes and reports its end by name.
105pub trait LinkCarrier: Send + std::fmt::Debug {
106    /// Which carrier holds this path.
107    fn carrier(&self) -> Carrier;
108    /// # Errors
109    /// A parked reader observed an unsolicited frame or a carrier end.
110    fn refuse_unsolicited(&self) -> Result<(), CarrierError> {
111        Ok(())
112    }
113    /// Sends one subject and its sealed bytes.
114    ///
115    /// # Errors
116    /// The frame was invalid or the carrier could not send it.
117    fn send(&mut self, subject: &str, sealed: &[u8]) -> Result<(), CarrierError>;
118    /// Receives the next sealed bytes, or the carrier's named end.
119    ///
120    /// # Errors
121    /// An I/O error, malformed frame, or unexpected frame/stream.
122    fn receive(&mut self) -> Result<CarrierReceive, CarrierError>;
123    /// The end this carrier observed, by name, once it has one: the peer's
124    /// close reason, `wire_eof`, or the failure that ended it. `None` while
125    /// the carrier is live.
126    fn ended(&self) -> Option<String> {
127        None
128    }
129}
130
131/// One application stream on the peer's TCP wire connection.
132#[derive(Debug)]
133pub struct WireCarrier {
134    socket: TcpStream,
135    stream_id: u32,
136    ended: Option<String>,
137    /// The first failure or end observed, by name: by a send or receive on
138    /// the owning thread, or by the reader thread as it reads the end.
139    failure: std::sync::Arc<std::sync::OnceLock<String>>,
140    life: Option<crate::link::LinkLife>,
141    reader: Option<reader::Reader>,
142}
143
144impl WireCarrier {
145    pub(crate) fn refuse_unsolicited(&self) -> Result<(), CarrierError> {
146        if let Some(reader) = &self.reader
147            && let Some(answer) = reader.try_receive()?
148        {
149            let error = match answer {
150                Ok(Some(_)) => CarrierError::UnexpectedFrame,
151                Ok(None) => CarrierError::Ended,
152                Err(error) => error,
153            };
154            self.failed(&error);
155            return Err(error);
156        }
157        Ok(())
158    }
159    pub(crate) fn install_life(&mut self, life: crate::link::LinkLife) -> Result<(), CarrierError> {
160        if self.life.is_some() {
161            return Err(CarrierError::Lifecycle(
162                "lifecycle already installed".to_owned(),
163            ));
164        }
165        self.life = Some(life.clone());
166        match reader::Reader::start(
167            &self.socket,
168            self.stream_id,
169            life,
170            std::sync::Arc::clone(&self.failure),
171        ) {
172            Ok(reader) => self.reader = Some(reader),
173            Err(error) => {
174                self.end_life(crate::link::LinkEnd::Refused(error.to_string()));
175                return Err(error);
176            }
177        }
178        Ok(())
179    }
180
181    pub(crate) const fn life(&self) -> Option<&crate::link::LinkLife> {
182        self.life.as_ref()
183    }
184
185    fn end_life(&self, reason: crate::link::LinkEnd) {
186        if let Some(life) = &self.life
187            && let Err(error) = life.end(reason)
188        {
189            tracing::error!(%error, "wire link end observation failed");
190        }
191    }
192
193    fn failed(&self, error: &CarrierError) {
194        self.failure.get_or_init(|| error.to_string());
195        self.end_life(
196            if matches!(error, CarrierError::Io(_) | CarrierError::Ended) {
197                crate::link::LinkEnd::PeerGone
198            } else {
199                crate::link::LinkEnd::Refused(error.to_string())
200            },
201        );
202    }
203    /// Duplicates the carrier socket solely to interrupt its blocking receive at stop.
204    ///
205    /// # Errors
206    /// The socket cannot be duplicated.
207    pub fn interrupt_handle(&self) -> Result<TcpStream, CarrierError> {
208        Ok(self.socket.try_clone()?)
209    }
210    /// Takes ownership of a connection whose link handshake has finished.
211    ///
212    /// # Errors
213    /// Stream zero is reserved for connection control.
214    pub fn new(socket: TcpStream, stream_id: u32) -> Result<Self, CarrierError> {
215        liminal::protocol::validate_stream(liminal::protocol::FrameType::LinkMessage, stream_id)?;
216        Ok(Self {
217            socket,
218            stream_id,
219            ended: None,
220            failure: std::sync::Arc::new(std::sync::OnceLock::new()),
221            life: None,
222            reader: None,
223        })
224    }
225}
226
227impl LinkCarrier for WireCarrier {
228    fn refuse_unsolicited(&self) -> Result<(), CarrierError> {
229        Self::refuse_unsolicited(self)
230    }
231    fn carrier(&self) -> Carrier {
232        Carrier::Wire
233    }
234    fn ended(&self) -> Option<String> {
235        self.ended
236            .clone()
237            .or_else(|| self.failure.get().cloned())
238            .or_else(|| {
239                self.life
240                    .as_ref()
241                    .and_then(|life| life.reason().ok().flatten())
242            })
243    }
244
245    fn send(&mut self, subject: &str, sealed: &[u8]) -> Result<(), CarrierError> {
246        if self.ended.is_some() {
247            return Err(CarrierError::Ended);
248        }
249        if let Some(life) = &self.life
250            && life.reason().map_err(CarrierError::Lifecycle)?.is_some()
251        {
252            return Err(CarrierError::Ended);
253        }
254        crate::auth_lys::SubjectBinding::new(
255            "wire".to_owned(),
256            subject.to_owned(),
257            subject.to_owned(),
258        )?;
259        let result = write_frame(
260            &mut self.socket,
261            &Frame::LinkMessage {
262                flags: 0,
263                stream_id: self.stream_id,
264                subject: subject.to_owned(),
265                sealed: sealed.to_vec(),
266            },
267        );
268        if let Err(error) = &result {
269            self.failed(error);
270        }
271        result
272    }
273
274    fn receive(&mut self) -> Result<CarrierReceive, CarrierError> {
275        if let Some(reason) = &self.ended {
276            return Ok(CarrierReceive::End {
277                reason: reason.clone(),
278            });
279        }
280        let answer = if let Some(reader) = &self.reader {
281            reader.receive()
282        } else {
283            read_frame(&mut self.socket)
284        };
285        let frame = answer.inspect_err(|error| self.failed(error))?;
286        let reason = match frame {
287            Some(Frame::PublishAck {
288                stream_id,
289                message_id,
290                durable_position,
291                ..
292            }) if stream_id == self.stream_id => {
293                return Ok(CarrierReceive::Acknowledged {
294                    message_id,
295                    durable_position,
296                });
297            }
298            Some(Frame::PublishError {
299                stream_id,
300                reason_code,
301                message,
302                ..
303            }) if stream_id == self.stream_id => {
304                return Ok(CarrierReceive::Refused {
305                    reason_code,
306                    reason: message,
307                });
308            }
309            Some(Frame::LinkMessage {
310                stream_id,
311                subject,
312                sealed,
313                ..
314            }) if stream_id == self.stream_id => {
315                crate::auth_lys::SubjectBinding::new(
316                    "wire".to_owned(),
317                    subject.clone(),
318                    subject.clone(),
319                )?;
320                return Ok(CarrierReceive::Sealed { subject, sealed });
321            }
322            Some(Frame::DisconnectReason { reason, .. }) => reason.name().to_owned(),
323            Some(Frame::Disconnect { .. }) => "peer_disconnect".to_owned(),
324            None => "wire_eof".to_owned(),
325            Some(_) => {
326                self.failed(&CarrierError::UnexpectedFrame);
327                return Err(CarrierError::UnexpectedFrame);
328            }
329        };
330        self.end_life(
331            if reason == "grant_revoked" || reason == "revocation_log_reset" {
332                crate::link::LinkEnd::Revoked
333            } else {
334                crate::link::LinkEnd::PeerGone
335            },
336        );
337        self.ended = Some(reason.clone());
338        Ok(CarrierReceive::End { reason })
339    }
340}
341
342impl Drop for WireCarrier {
343    fn drop(&mut self) {
344        self.end_life(crate::link::LinkEnd::Shutdown);
345        if let Some(reader) = self.reader.take() {
346            reader.close(&self.socket);
347        }
348    }
349}
350
351/// The same bound is enforced before encoding and before allocating a peer's body.
352pub(crate) const WIRE_FRAME_BOUND: usize = LimitsConfig::DEFAULT_MAX_CONNECTION_INBOX_BYTES;
353
354pub(crate) fn write_frame(socket: &mut TcpStream, frame: &Frame) -> Result<(), CarrierError> {
355    let bytes = codec::encoded_len(frame)?;
356    if bytes > WIRE_FRAME_BOUND {
357        return Err(CarrierError::TooLarge {
358            bytes,
359            bound: WIRE_FRAME_BOUND,
360        });
361    }
362    let mut encoded = vec![0; bytes];
363    let length = codec::encode(frame, &mut encoded)?;
364    socket.write_all(&encoded[..length])?;
365    Ok(())
366}
367
368pub(crate) fn read_frame(socket: &mut TcpStream) -> Result<Option<Frame>, CarrierError> {
369    let mut header = [0; FrameHeader::WIRE_LEN];
370    // Only EOF before a header is a clean end; an incomplete header/body is an error.
371    loop {
372        match socket.read(&mut header[..1]) {
373            Ok(0) => return Ok(None),
374            Ok(_) => break,
375            Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
376            Err(error) => return Err(error.into()),
377        }
378    }
379    socket.read_exact(&mut header[1..])?;
380    let payload = u32::from_be_bytes([header[6], header[7], header[8], header[9]]);
381    let bytes = usize::try_from(payload)
382        .ok()
383        .and_then(|length| length.checked_add(header.len()))
384        .ok_or(CarrierError::TooLarge {
385            bytes: usize::MAX,
386            bound: WIRE_FRAME_BOUND,
387        })?;
388    if bytes > WIRE_FRAME_BOUND {
389        return Err(CarrierError::TooLarge {
390            bytes,
391            bound: WIRE_FRAME_BOUND,
392        });
393    }
394    let mut encoded = vec![0; bytes];
395    encoded[..header.len()].copy_from_slice(&header);
396    socket.read_exact(&mut encoded[header.len()..])?;
397    let (frame, _) = codec::decode(&encoded)?;
398    Ok(Some(frame))
399}
liminal_server/link_carrier.rs

liminal_server/
link_carrier.rs
