radicle_protocol/wire/
message.rs

1use std::{mem, net};
2
3use bytes::Buf;
4use bytes::BufMut;
5#[cfg(feature = "i2p")]
6use cypheraddr::i2p;
7#[cfg(feature = "tor")]
8use cypheraddr::tor;
9use cypheraddr::{HostName, NetAddr};
10use radicle::crypto::Signature;
11use radicle::git::Oid;
12use radicle::identity::RepoId;
13use radicle::node::Address;
14use radicle::node::NodeId;
15use radicle::node::Timestamp;
16use radicle::node::address::AddressType;
17
18use crate::bounded::BoundedVec;
19use crate::service::filter::Filter;
20use crate::service::message::*;
21use crate::wire;
22
23/// Message type.
24#[repr(u16)]
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum MessageType {
27    NodeAnnouncement = 2,
28    InventoryAnnouncement = 4,
29    RefsAnnouncement = 6,
30    Subscribe = 8,
31    Ping = 10,
32    Pong = 12,
33    Info = 14,
34}
35
36impl From<MessageType> for u16 {
37    fn from(other: MessageType) -> Self {
38        other as u16
39    }
40}
41
42impl TryFrom<u16> for MessageType {
43    type Error = u16;
44
45    fn try_from(other: u16) -> Result<Self, Self::Error> {
46        match other {
47            2 => Ok(MessageType::NodeAnnouncement),
48            4 => Ok(MessageType::InventoryAnnouncement),
49            6 => Ok(MessageType::RefsAnnouncement),
50            8 => Ok(MessageType::Subscribe),
51            10 => Ok(MessageType::Ping),
52            12 => Ok(MessageType::Pong),
53            14 => Ok(MessageType::Info),
54            _ => Err(other),
55        }
56    }
57}
58
59impl Message {
60    /// The maximum supported message size in bytes.
61    pub const MAX_SIZE: wire::Size =
62        wire::Size::MAX - (mem::size_of::<MessageType>() as wire::Size);
63
64    pub fn type_id(&self) -> u16 {
65        match self {
66            Self::Subscribe { .. } => MessageType::Subscribe,
67            Self::Announcement(Announcement { message, .. }) => match message {
68                AnnouncementMessage::Node(_) => MessageType::NodeAnnouncement,
69                AnnouncementMessage::Inventory(_) => MessageType::InventoryAnnouncement,
70                AnnouncementMessage::Refs(_) => MessageType::RefsAnnouncement,
71            },
72            Self::Info(_) => MessageType::Info,
73            Self::Ping { .. } => MessageType::Ping,
74            Self::Pong { .. } => MessageType::Pong,
75        }
76        .into()
77    }
78}
79
80impl wire::Encode for AnnouncementMessage {
81    fn encode(&self, buf: &mut impl BufMut) {
82        match self {
83            Self::Node(ann) => ann.encode(buf),
84            Self::Inventory(ann) => ann.encode(buf),
85            Self::Refs(ann) => ann.encode(buf),
86        }
87    }
88}
89
90impl wire::Encode for RefsAnnouncement {
91    fn encode(&self, buf: &mut impl BufMut) {
92        self.rid.encode(buf);
93        self.refs.encode(buf);
94        self.timestamp.encode(buf);
95    }
96}
97
98impl wire::Decode for RefsAnnouncement {
99    fn decode(buf: &mut impl Buf) -> Result<Self, wire::Error> {
100        let rid = RepoId::decode(buf)?;
101        let refs = BoundedVec::<_, REF_REMOTE_LIMIT>::decode(buf)?;
102        let timestamp = Timestamp::decode(buf)?;
103
104        Ok(Self {
105            rid,
106            refs,
107            timestamp,
108        })
109    }
110}
111
112impl wire::Encode for InventoryAnnouncement {
113    fn encode(&self, buf: &mut impl BufMut) {
114        self.inventory.encode(buf);
115        self.timestamp.encode(buf);
116    }
117}
118
119impl wire::Decode for InventoryAnnouncement {
120    fn decode(buf: &mut impl Buf) -> Result<Self, wire::Error> {
121        let inventory = BoundedVec::decode(buf)?;
122        let timestamp = Timestamp::decode(buf)?;
123
124        Ok(Self {
125            inventory,
126            timestamp,
127        })
128    }
129}
130
131/// The type tracking the different variants of [`Info`] for encoding and
132/// decoding purposes.
133#[repr(u8)]
134#[derive(Debug, Clone, Copy, PartialEq, Eq)]
135pub enum InfoType {
136    RefsAlreadySynced = 1,
137}
138
139impl From<InfoType> for u16 {
140    fn from(other: InfoType) -> Self {
141        other as u16
142    }
143}
144
145impl TryFrom<u16> for InfoType {
146    type Error = u16;
147
148    fn try_from(other: u16) -> Result<Self, Self::Error> {
149        match other {
150            1 => Ok(Self::RefsAlreadySynced),
151            n => Err(n),
152        }
153    }
154}
155
156impl From<Info> for InfoType {
157    fn from(info: Info) -> Self {
158        (&info).into()
159    }
160}
161
162impl From<&Info> for InfoType {
163    fn from(info: &Info) -> Self {
164        match info {
165            Info::RefsAlreadySynced { .. } => Self::RefsAlreadySynced,
166        }
167    }
168}
169
170impl wire::Encode for Info {
171    fn encode(&self, buf: &mut impl BufMut) {
172        u16::from(InfoType::from(self)).encode(buf);
173        match self {
174            Info::RefsAlreadySynced { rid, at } => {
175                rid.encode(buf);
176                at.encode(buf);
177            }
178        }
179    }
180}
181
182impl wire::Decode for Info {
183    fn decode(buf: &mut impl Buf) -> Result<Self, wire::Error> {
184        let info_type = buf.try_get_u16()?;
185
186        match InfoType::try_from(info_type) {
187            Ok(InfoType::RefsAlreadySynced) => {
188                let rid = RepoId::decode(buf)?;
189                let at = Oid::decode(buf)?;
190
191                Ok(Self::RefsAlreadySynced { rid, at })
192            }
193            Err(other) => Err(wire::Invalid::InfoMessageType { actual: other }.into()),
194        }
195    }
196}
197
198impl wire::Encode for Message {
199    fn encode(&self, buf: &mut impl BufMut) {
200        let buf = &mut buf.limit(wire::Size::MAX as usize);
201
202        self.type_id().encode(buf);
203
204        match self {
205            Self::Subscribe(Subscribe {
206                filter,
207                since,
208                until,
209            }) => {
210                filter.encode(buf);
211                since.encode(buf);
212                until.encode(buf);
213            }
214            Self::Announcement(Announcement {
215                node,
216                message,
217                signature,
218            }) => {
219                node.encode(buf);
220                signature.encode(buf);
221                message.encode(buf);
222            }
223            Self::Info(info) => {
224                info.encode(buf);
225            }
226            Self::Ping(Ping { ponglen, zeroes }) => {
227                ponglen.encode(buf);
228                zeroes.encode(buf);
229            }
230            Self::Pong { zeroes } => {
231                zeroes.encode(buf);
232            }
233        }
234    }
235}
236
237impl wire::Decode for Message {
238    fn decode(buf: &mut impl Buf) -> Result<Self, wire::Error> {
239        let type_id = buf.try_get_u16()?;
240
241        match MessageType::try_from(type_id) {
242            Ok(MessageType::Subscribe) => {
243                let filter = Filter::decode(buf)?;
244                let since = Timestamp::decode(buf)?;
245                let until = Timestamp::decode(buf)?;
246
247                Ok(Self::Subscribe(Subscribe {
248                    filter,
249                    since,
250                    until,
251                }))
252            }
253            Ok(MessageType::NodeAnnouncement) => {
254                let node = NodeId::decode(buf)?;
255                let signature = Signature::decode(buf)?;
256                let message = NodeAnnouncement::decode(buf)?.into();
257
258                Ok(Announcement {
259                    node,
260                    message,
261                    signature,
262                }
263                .into())
264            }
265            Ok(MessageType::InventoryAnnouncement) => {
266                let node = NodeId::decode(buf)?;
267                let signature = Signature::decode(buf)?;
268                let message = InventoryAnnouncement::decode(buf)?.into();
269
270                Ok(Announcement {
271                    node,
272                    message,
273                    signature,
274                }
275                .into())
276            }
277            Ok(MessageType::RefsAnnouncement) => {
278                let node = NodeId::decode(buf)?;
279                let signature = Signature::decode(buf)?;
280                let message = RefsAnnouncement::decode(buf)?.into();
281
282                Ok(Announcement {
283                    node,
284                    message,
285                    signature,
286                }
287                .into())
288            }
289            Ok(MessageType::Info) => {
290                let info = Info::decode(buf)?;
291                Ok(Self::Info(info))
292            }
293            Ok(MessageType::Ping) => {
294                let ponglen = u16::decode(buf)?;
295                let zeroes = ZeroBytes::decode(buf)?;
296                Ok(Self::Ping(Ping { ponglen, zeroes }))
297            }
298            Ok(MessageType::Pong) => {
299                let zeroes = ZeroBytes::decode(buf)?;
300                Ok(Self::Pong { zeroes })
301            }
302            Err(other) => Err(wire::Invalid::MessageType { actual: other }.into()),
303        }
304    }
305}
306
307impl wire::Encode for Address {
308    fn encode(&self, buf: &mut impl BufMut) {
309        match self.host {
310            HostName::Ip(net::IpAddr::V4(ip)) => {
311                u8::from(AddressType::Ipv4).encode(buf);
312                ip.octets().encode(buf);
313            }
314            HostName::Ip(net::IpAddr::V6(ip)) => {
315                u8::from(AddressType::Ipv6).encode(buf);
316                ip.octets().encode(buf);
317            }
318            HostName::Dns(ref dns) => {
319                u8::from(AddressType::Dns).encode(buf);
320                dns.encode(buf);
321            }
322            #[cfg(feature = "tor")]
323            HostName::Tor(addr) => {
324                u8::from(AddressType::Onion).encode(buf);
325                addr.encode(buf);
326            }
327            #[cfg(feature = "i2p")]
328            HostName::I2p(ref addr) => {
329                u8::from(AddressType::I2p).encode(buf);
330                addr.encode(buf);
331            }
332            _ => {
333                unimplemented!(
334                    "Encoding not defined for addresses of the same type as the following: {:?}",
335                    self.host
336                );
337            }
338        }
339        self.port().encode(buf);
340    }
341}
342
343impl wire::Decode for Address {
344    fn decode(buf: &mut impl Buf) -> Result<Self, wire::Error> {
345        let addrtype = buf.try_get_u8()?;
346
347        let host = match AddressType::try_from(addrtype) {
348            Ok(AddressType::Ipv4) => {
349                let octets: [u8; 4] = wire::Decode::decode(buf)?;
350                let ip = net::Ipv4Addr::from(octets);
351
352                HostName::Ip(net::IpAddr::V4(ip))
353            }
354            Ok(AddressType::Ipv6) => {
355                let octets: [u8; 16] = wire::Decode::decode(buf)?;
356                let ip = net::Ipv6Addr::from(octets);
357
358                HostName::Ip(net::IpAddr::V6(ip))
359            }
360            Ok(AddressType::Dns) => {
361                let dns: String = wire::Decode::decode(buf)?;
362
363                HostName::Dns(dns)
364            }
365            #[cfg(feature = "tor")]
366            Ok(AddressType::Onion) => {
367                let onion: tor::OnionAddrV3 = wire::Decode::decode(buf)?;
368
369                HostName::Tor(onion)
370            }
371            #[cfg(feature = "i2p")]
372            Ok(AddressType::I2p) => {
373                let i2p: i2p::I2pAddr = wire::Decode::decode(buf)?;
374
375                HostName::I2p(i2p)
376            }
377            Ok(unknown) => {
378                return Err(wire::Invalid::AddressType {
379                    actual: unknown.into(),
380                }
381                .into());
382            }
383            Err(other) => return Err(wire::Invalid::AddressType { actual: other }.into()),
384        };
385        let port = u16::decode(buf)?;
386
387        Ok(Self::from(NetAddr { host, port }))
388    }
389}
390
391impl wire::Encode for ZeroBytes {
392    fn encode(&self, buf: &mut impl BufMut) {
393        (self.len() as u16).encode(buf);
394        buf.put_bytes(0u8, self.len());
395    }
396}
397
398impl wire::Decode for ZeroBytes {
399    fn decode(buf: &mut impl Buf) -> Result<Self, wire::Error> {
400        let zeroes = u16::decode(buf)?;
401        for _ in 0..zeroes {
402            _ = u8::decode(buf)?;
403        }
404        Ok(ZeroBytes::new(zeroes))
405    }
406}
407
408#[cfg(test)]
409mod tests {
410    use qcheck_macros::quickcheck;
411    use radicle::node::UserAgent;
412    use radicle::storage::refs::RefsAt;
413    use radicle::test::arbitrary;
414
415    use crate::deserializer::Deserializer;
416    use crate::prop_roundtrip;
417    use crate::wire::{Encode as _, roundtrip};
418
419    use super::*;
420
421    prop_roundtrip!(Address);
422    prop_roundtrip!(Message);
423
424    #[test]
425    fn test_refs_ann_max_size() {
426        let signer = radicle::crypto::SigningKey::mock(235);
427        let refs: [RefsAt; REF_REMOTE_LIMIT] = arbitrary::r#gen(1);
428        let ann = AnnouncementMessage::Refs(RefsAnnouncement {
429            rid: arbitrary::r#gen(1),
430            refs: BoundedVec::collect_from(&mut refs.into_iter()),
431            timestamp: arbitrary::r#gen(1),
432        });
433        let ann = ann.signed(&signer);
434        let msg = Message::Announcement(ann);
435        let data = msg.encode_to_vec();
436
437        assert!(data.len() < wire::Size::MAX as usize);
438    }
439
440    #[test]
441    fn test_inv_ann_max_size() {
442        let signer = radicle::crypto::SigningKey::mock(147);
443        let inv: [RepoId; INVENTORY_LIMIT] = arbitrary::r#gen(1);
444        let ann = AnnouncementMessage::Inventory(InventoryAnnouncement {
445            inventory: BoundedVec::collect_from(&mut inv.into_iter()),
446            timestamp: arbitrary::r#gen(1),
447        });
448        let ann = ann.signed(&signer);
449        let msg = Message::Announcement(ann);
450        let data = msg.encode_to_vec();
451
452        assert!(data.len() < wire::Size::MAX as usize);
453    }
454
455    #[test]
456    fn test_node_ann_max_size() {
457        let signer = radicle::crypto::SigningKey::mock(247);
458        let addrs: [Address; ADDRESS_LIMIT] = arbitrary::r#gen(1);
459        let alias = ['@'; radicle::node::MAX_ALIAS_LENGTH];
460        let ann = AnnouncementMessage::Node(NodeAnnouncement {
461            version: 1,
462            features: Default::default(),
463            alias: radicle::node::Alias::new(String::from_iter(alias)),
464            addresses: BoundedVec::collect_from(&mut addrs.into_iter()),
465            timestamp: arbitrary::r#gen(1),
466            nonce: u64::MAX,
467            agent: UserAgent::default(),
468        });
469        let ann = ann.signed(&signer);
470        let msg = Message::Announcement(ann);
471        let data = msg.encode_to_vec();
472
473        assert!(data.len() < wire::Size::MAX as usize);
474    }
475
476    #[test]
477    fn test_pingpong_encode_max_size() {
478        Message::Ping(Ping {
479            ponglen: 0,
480            zeroes: ZeroBytes::new(Ping::MAX_PING_ZEROES),
481        })
482        .encode_to_vec();
483
484        (Message::Pong {
485            zeroes: ZeroBytes::new(Ping::MAX_PONG_ZEROES),
486        })
487        .encode_to_vec();
488    }
489
490    #[test]
491    #[should_panic(expected = "advance out of bounds")]
492    fn test_ping_encode_size_overflow() {
493        Message::Ping(Ping {
494            ponglen: 0,
495            zeroes: ZeroBytes::new(Ping::MAX_PING_ZEROES + 1),
496        })
497        .encode_to_vec();
498    }
499
500    #[test]
501    #[should_panic(expected = "advance out of bounds")]
502    fn test_pong_encode_size_overflow() {
503        Message::Pong {
504            zeroes: ZeroBytes::new(Ping::MAX_PONG_ZEROES + 1),
505        }
506        .encode_to_vec();
507    }
508
509    #[test]
510    fn prop_message_decoder() {
511        fn property(items: Vec<Message>) {
512            let mut decoder = Deserializer::<1048576, Message>::new(8);
513
514            for item in &items {
515                item.encode(&mut decoder);
516            }
517            for item in items {
518                assert_eq!(decoder.next().unwrap().unwrap(), item);
519            }
520        }
521
522        qcheck::QuickCheck::new()
523            .r#gen(qcheck::Gen::new(16))
524            .quickcheck(property as fn(items: Vec<Message>));
525    }
526
527    #[quickcheck]
528    fn prop_zero_bytes_encode_decode(zeroes: wire::Size) -> qcheck::TestResult {
529        if zeroes > Ping::MAX_PING_ZEROES {
530            return qcheck::TestResult::discard();
531        }
532
533        roundtrip(ZeroBytes::new(zeroes));
534
535        qcheck::TestResult::passed()
536    }
537}
radicle_protocol/wire/message.rs

radicle_protocol/wire/
message.rs
