radicle_protocol/wire/
frame.rs

1//! Framing protocol.
2#![warn(clippy::missing_docs_in_private_items)]
3use std::{fmt, io};
4
5use bytes::{Buf, BufMut};
6use radicle::node::Link;
7
8use crate::service::Message;
9use crate::{PROTOCOL_VERSION, wire, wire::varint, wire::varint::VarInt};
10
11/// Protocol version strings all start with the magic sequence `rad`, followed
12/// by a version number.
13pub const PROTOCOL_VERSION_STRING: Version = Version([b'r', b'a', b'd', PROTOCOL_VERSION]);
14
15/// Protocol version.
16#[derive(Debug, PartialEq, Eq)]
17pub struct Version([u8; 4]);
18
19impl Version {
20    /// Version number.
21    pub fn number(&self) -> u8 {
22        self.0[3]
23    }
24}
25
26impl wire::Encode for Version {
27    fn encode(&self, buf: &mut impl BufMut) {
28        buf.put_slice(&PROTOCOL_VERSION_STRING.0);
29    }
30}
31
32impl wire::Decode for Version {
33    fn decode(buf: &mut impl Buf) -> Result<Self, wire::Error> {
34        let mut version = [0u8; 4];
35
36        buf.try_copy_to_slice(&mut version[..])?;
37
38        if version != PROTOCOL_VERSION_STRING.0 {
39            return Err(wire::Invalid::ProtocolVersion { actual: version }.into());
40        }
41        Ok(Self(version))
42    }
43}
44
45/// Identifies a (multiplexed) stream.
46///
47/// Stream IDs are variable-length integers with the least significant 3 bits
48/// denoting the stream type and initiator.
49///
50/// The first bit denotes the initiator (outbound or inbound), while the second
51/// and third bit denote the stream type. See `StreamKind`.
52///
53/// In a situation where Alice connects to Bob, Alice will have the initiator
54/// bit set to `1` for all streams she creates, while Bob will have it set to `0`.
55///
56/// This ensures that Stream IDs never collide.
57/// Additionally, Stream IDs must never be reused within a connection.
58///
59/// +=======+==================================+
60/// | Bits  | Stream Type                      |
61/// +=======+==================================+
62/// | 0b000 | Outbound Control stream          |
63/// +-------+----------------------------------+
64/// | 0b001 | Inbound Control stream           |
65/// +-------+----------------------------------+
66/// | 0b010 | Outbound Gossip stream           |
67/// +-------+----------------------------------+
68/// | 0b011 | Inbound Gossip stream            |
69/// +-------+----------------------------------+
70/// | 0b100 | Outbound Git stream              |
71/// +-------+----------------------------------+
72/// | 0b101 | Inbound Git stream               |
73/// +-------+----------------------------------+
74///
75#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
76pub struct StreamId(VarInt);
77
78impl StreamId {
79    /// Get the initiator of this stream.
80    pub fn link(&self) -> Link {
81        let n = *self.0;
82        if 0b1 & n == 0 {
83            Link::Outbound
84        } else {
85            Link::Inbound
86        }
87    }
88
89    /// Get the kind of stream this is.
90    pub fn kind(&self) -> Result<StreamType, u8> {
91        let id = *self.0;
92        let kind = ((id >> 1) & 0b11) as u8;
93
94        StreamType::try_from(kind)
95    }
96
97    /// Create a control identifier.
98    pub fn control(link: Link) -> Self {
99        let link = if link.is_outbound() { 0 } else { 1 };
100        Self(VarInt::from(((u8::from(StreamType::Control)) << 1) | link))
101    }
102
103    /// Create a gossip identifier.
104    pub fn gossip(link: Link) -> Self {
105        let link = if link.is_outbound() { 0 } else { 1 };
106        Self(VarInt::from((u8::from(StreamType::Gossip) << 1) | link))
107    }
108
109    /// Create a git identifier.
110    pub fn git(link: Link) -> Self {
111        let link = if link.is_outbound() { 0 } else { 1 };
112        Self(VarInt::from((u8::from(StreamType::Git) << 1) | link))
113    }
114
115    /// Get the nth identifier while preserving the stream type and initiator.
116    pub fn nth(self, n: u64) -> Result<Self, varint::BoundsExceeded> {
117        let id = *self.0 + (n << 3);
118        VarInt::new(id).map(Self)
119    }
120}
121
122impl From<StreamId> for u64 {
123    fn from(value: StreamId) -> Self {
124        *value.0
125    }
126}
127
128impl From<StreamId> for VarInt {
129    fn from(value: StreamId) -> Self {
130        value.0
131    }
132}
133
134impl fmt::Display for StreamId {
135    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
136        write!(f, "{}", *self.0)
137    }
138}
139
140impl wire::Decode for StreamId {
141    fn decode(buf: &mut impl Buf) -> Result<Self, wire::Error> {
142        let id = VarInt::decode(buf)?;
143        Ok(Self(id))
144    }
145}
146
147impl wire::Encode for StreamId {
148    fn encode(&self, buf: &mut impl BufMut) {
149        self.0.encode(buf)
150    }
151}
152
153/// Type of stream.
154#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
155#[repr(u8)]
156pub enum StreamType {
157    /// Control stream, used to open and close streams.
158    Control = 0b00,
159    /// Gossip stream, used to exchange messages.
160    Gossip = 0b01,
161    /// Git stream, used for replication.
162    Git = 0b10,
163}
164
165impl TryFrom<u8> for StreamType {
166    type Error = u8;
167
168    fn try_from(value: u8) -> Result<Self, Self::Error> {
169        match value {
170            0b00 => Ok(StreamType::Control),
171            0b01 => Ok(StreamType::Gossip),
172            0b10 => Ok(StreamType::Git),
173            n => Err(n),
174        }
175    }
176}
177
178impl From<StreamType> for u8 {
179    fn from(value: StreamType) -> Self {
180        value as u8
181    }
182}
183
184/// Protocol frame.
185///
186/// ```text
187///  0                   1                   2                   3
188///  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
189/// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
190/// |      'r'      |      'a'      |      'd'      |      0x1      | Version
191/// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
192/// |                                                       |T T T I| Stream ID
193/// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
194/// |                                                              …| Data (variable size)
195/// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
196/// ```
197///
198/// Note that the last four bits of the Stream ID encode the Stream Type
199/// and Initiator.
200/// The first three bits of the last four bits encode the Stream Type,
201/// while the last bit encodes the Initiator.
202#[derive(Debug, PartialEq, Eq)]
203pub struct Frame<M = Message> {
204    /// The protocol version.
205    pub version: Version,
206    /// The stream identifier.
207    pub stream: StreamId,
208    /// The frame payload.
209    pub data: FrameData<M>,
210}
211
212impl<M> Frame<M> {
213    /// Create a 'git' protocol frame.
214    pub fn git(stream: StreamId, data: Vec<u8>) -> Self {
215        Self {
216            version: PROTOCOL_VERSION_STRING,
217            stream,
218            data: FrameData::Git(data),
219        }
220    }
221
222    /// Create a 'control' protocol frame.
223    pub fn control(link: Link, ctrl: Control) -> Self {
224        Self {
225            version: PROTOCOL_VERSION_STRING,
226            stream: StreamId::control(link),
227            data: FrameData::Control(ctrl),
228        }
229    }
230
231    /// Create a 'gossip' protocol frame.
232    pub fn gossip(link: Link, msg: M) -> Self {
233        Self {
234            version: PROTOCOL_VERSION_STRING,
235            stream: StreamId::gossip(link),
236            data: FrameData::Gossip(msg),
237        }
238    }
239}
240
241/// Frame payload.
242#[derive(Debug, PartialEq, Eq)]
243pub enum FrameData<M> {
244    /// Control frame payload.
245    Control(Control),
246    /// Gossip frame payload.
247    Gossip(M),
248    /// Git frame payload. May contain packet-lines as well as packfile data.
249    Git(Vec<u8>),
250}
251
252/// A control message sent over a control stream.
253#[derive(Debug, PartialEq, Eq)]
254pub enum Control {
255    /// Open a new stream.
256    Open {
257        /// The stream to open.
258        stream: StreamId,
259    },
260    /// Close an existing stream.
261    Close {
262        /// The stream to close.
263        stream: StreamId,
264    },
265    /// Signal an end-of-file. This can be used to simulate connections terminating
266    /// without having to close the connection. These control messages are turned into
267    /// [`io::ErrorKind::UnexpectedEof`] errors on read.
268    Eof {
269        /// The stream to send an EOF on.
270        stream: StreamId,
271    },
272}
273
274/// Type of control message.
275#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
276#[repr(u8)]
277pub enum ControlType {
278    /// Control open byte.
279    Open = 0,
280    /// Control close byte.
281    Close = 1,
282    /// Control EOF byte.
283    Eof = 2,
284}
285
286impl TryFrom<u8> for ControlType {
287    type Error = u8;
288
289    fn try_from(value: u8) -> Result<Self, Self::Error> {
290        match value {
291            0b00 => Ok(ControlType::Open),
292            0b01 => Ok(ControlType::Close),
293            0b10 => Ok(ControlType::Eof),
294            n => Err(n),
295        }
296    }
297}
298
299impl From<ControlType> for u8 {
300    fn from(value: ControlType) -> Self {
301        value as u8
302    }
303}
304
305impl wire::Decode for Control {
306    fn decode(buf: &mut impl Buf) -> Result<Self, wire::Error> {
307        match ControlType::try_from(u8::decode(buf)?) {
308            Ok(ControlType::Open) => Ok(Control::Open {
309                stream: StreamId::decode(buf)?,
310            }),
311            Ok(ControlType::Close) => Ok(Control::Close {
312                stream: StreamId::decode(buf)?,
313            }),
314            Ok(ControlType::Eof) => Ok(Control::Eof {
315                stream: StreamId::decode(buf)?,
316            }),
317            Err(other) => Err(wire::Invalid::ControlType { actual: other }.into()),
318        }
319    }
320}
321
322impl wire::Encode for Control {
323    fn encode(&self, buf: &mut impl BufMut) {
324        match self {
325            Self::Open { stream: id } => {
326                u8::from(ControlType::Open).encode(buf);
327                id.encode(buf);
328            }
329            Self::Eof { stream: id } => {
330                u8::from(ControlType::Eof).encode(buf);
331                id.encode(buf);
332            }
333            Self::Close { stream: id } => {
334                u8::from(ControlType::Close).encode(buf);
335                id.encode(buf);
336            }
337        }
338    }
339}
340
341impl<M: wire::Decode> wire::Decode for Frame<M> {
342    fn decode(buf: &mut impl Buf) -> Result<Self, wire::Error> {
343        let version = Version::decode(buf)?;
344        if version.number() != PROTOCOL_VERSION {
345            return Err(wire::Invalid::ProtocolVersionUnsupported {
346                actual: version.number(),
347            }
348            .into());
349        }
350        let stream = StreamId::decode(buf)?;
351
352        match stream.kind() {
353            Ok(StreamType::Control) => {
354                let ctrl = Control::decode(buf)?;
355                let frame = Frame {
356                    version,
357                    stream,
358                    data: FrameData::Control(ctrl),
359                };
360                Ok(frame)
361            }
362            Ok(StreamType::Gossip) => {
363                let data = varint::payload::decode(buf)?;
364                let mut cursor = io::Cursor::new(data);
365                let msg = M::decode(&mut cursor)?;
366                let frame = Frame {
367                    version,
368                    stream,
369                    data: FrameData::Gossip(msg),
370                };
371
372                // Nb. If there is data after the `Message` that is not decoded,
373                // it is simply dropped here.
374
375                Ok(frame)
376            }
377            Ok(StreamType::Git) => {
378                let data = varint::payload::decode(buf)?;
379                Ok(Frame::git(stream, data))
380            }
381            Err(n) => Err(wire::Invalid::StreamType { actual: n }.into()),
382        }
383    }
384}
385
386impl<M: wire::Encode> wire::Encode for Frame<M> {
387    fn encode(&self, buf: &mut impl BufMut) {
388        self.version.encode(buf);
389        self.stream.encode(buf);
390        match &self.data {
391            FrameData::Control(ctrl) => ctrl.encode(buf),
392            FrameData::Git(data) => varint::payload::encode(data, buf),
393            FrameData::Gossip(msg) => varint::payload::encode(&msg.encode_to_vec(), buf),
394        }
395    }
396}
397
398#[cfg(test)]
399mod test {
400    use super::*;
401
402    #[test]
403    fn test_stream_id() {
404        assert_eq!(StreamId(VarInt(0b000)).kind().unwrap(), StreamType::Control);
405        assert_eq!(StreamId(VarInt(0b010)).kind().unwrap(), StreamType::Gossip);
406        assert_eq!(StreamId(VarInt(0b100)).kind().unwrap(), StreamType::Git);
407        assert_eq!(StreamId(VarInt(0b001)).link(), Link::Inbound);
408        assert_eq!(StreamId(VarInt(0b000)).link(), Link::Outbound);
409        assert_eq!(StreamId(VarInt(0b101)).link(), Link::Inbound);
410        assert_eq!(StreamId(VarInt(0b100)).link(), Link::Outbound);
411
412        assert_eq!(StreamId::git(Link::Outbound), StreamId(VarInt(0b100)));
413        assert_eq!(StreamId::control(Link::Outbound), StreamId(VarInt(0b000)));
414        assert_eq!(StreamId::gossip(Link::Outbound), StreamId(VarInt(0b010)));
415
416        assert_eq!(StreamId::git(Link::Inbound), StreamId(VarInt(0b101)));
417        assert_eq!(StreamId::control(Link::Inbound), StreamId(VarInt(0b001)));
418        assert_eq!(StreamId::gossip(Link::Inbound), StreamId(VarInt(0b011)));
419    }
420
421    #[test]
422    fn test_encode_git_large() {
423        use wire::Encode as _;
424
425        let size = u16::MAX as usize * 3;
426        assert!(
427            size > (wire::Size::MAX as usize * 2),
428            "we want to test sizes that are way larger than any gossip message"
429        );
430
431        let a_lot_of_data = vec![0u8; size];
432
433        let frame: Frame<Message> = Frame::git(StreamId(0u8.into()), a_lot_of_data);
434
435        // In previous versions since 3c5668e this would panic.
436        let bytes = frame.encode_to_vec();
437
438        assert!(
439            bytes.len() > wire::Size::MAX as usize * 2,
440            "just making sure that whatever was encoded is still quite large"
441        );
442    }
443}
radicle_protocol/wire/frame.rs

radicle_protocol/wire/
frame.rs
