protocol: Revisit encoding and decoding — heartwood

Encoding

In 3c5668edd22ae4b9a085220d6be552f944ccb038 I mistakenly moved the check the length of an encoded wire::Message from its impl wire::Encode to the fn serialize, so that it would not apply only to messages (as originally intended), but to any argument of serialize.

This is not a problem for gossip messages, but catastrophic for Git streams, which tend to send a lot of data.

The fix is simple: Move the check back into impl Encode for Message.

What made this harder to spot for me was the multitude of concepts and terms involved in encoding: What’s serializing vs. encoding? Why is there this odd Frame::to_bytes?

So I decided to clean up a bit. I removed fn serialize so that the logic stays close to the data, and instead of it provide fn encode_vec in trait Encode so that we don’t add up with one-offs like Frame::to_bytes again. Also we make use of encode_vec in the tests.

There is test_encode_git_large will panic if we ever regress.

Decoding

Remove fn deserialize and instead provide a trait fn decode_exact for testing.

Refactor errors, so that we can now cleanly distinguish between Error::UnexpectedEnd which means that decoding might work given more data, and Error::Invalid, which means that no amount of additional data will decode.

While at it, cleanup the variants of Error::Invalid, and provide ControlType.

                         stream: task.stream,
                     },
                 );
                 self.actions.push_back(Action::Send(fd, frame.to_bytes()));
                 self.actions.push_back(Action::Send(fd, frame.encode_vec()));
             }
         } else {
             // If the peer disconnected, we'll get here, but we still want to let the service know

                 ChannelEvent::Eof => Frame::control(*link, frame::Control::Eof { stream }),
             };
             self.actions
                 .push_back(reactor::Action::Send(fd, frame.to_bytes()));
                 .push_back(reactor::Action::Send(fd, frame.encode_vec()));
         }
     }

                     self.actions.push_back(Action::Send(
                         fd,
                         Frame::<service::Message>::control(link, frame::Control::Open { stream })
                             .to_bytes(),
                             .encode_vec(),
                     ));
                 }
             }

         frame::StreamId::gossip(Link::Outbound).encode(&mut stream);
         // Serialize gossip message with some extension fields.
         let mut gossip = wire::serialize(&pong);
         let mut gossip = pong.encode_vec();
         String::from("extra").encode(&mut gossip);
 u8.encode(&mut gossip);

     }
     /// Decode and return the next message. Returns [`None`] if nothing was decoded.
     pub fn deserialize_next(&mut self) -> Result<Option<D>, wire::Error> {
     pub fn deserialize_next(&mut self) -> Result<Option<D>, wire::Invalid> {
         let mut reader = io::Cursor::new(self.unparsed.as_slice());
         match D::decode(&mut reader) {

                 Ok(Some(msg))
             }
             Err(wire::Error::UnexpectedEnd { .. }) => Ok(None),
             Err(err) => Err(err),
             Err(wire::Error::Invalid(err)) => Err(err),
         }
     }

 }
 impl<const B: usize, D: wire::Decode> Iterator for Deserializer<B, D> {
     type Item = Result<D, wire::Error>;
     type Item = Result<D, wire::Invalid>;
     fn next(&mut self) -> Option<Self::Item> {
         self.deserialize_next().transpose()

     Announcement, AnnouncementMessage, InventoryAnnouncement, NodeAnnouncement, RefsAnnouncement,
 };
 use crate::wire;
 use crate::wire::Decode;
 use crate::wire::{Decode as _, Encode as _};
 use radicle::node::Database;
 use radicle::node::NodeId;
 use radicle::prelude::Timestamp;

             AnnouncementMessage::Node(msg) => {
                 stmt.bind((2, sql::Value::String(String::new())))?;
                 stmt.bind((3, &GossipType::Node))?;
                 stmt.bind((4, msg))?;
                 stmt.bind((4, &msg.encode_vec()[..]))?;
             }
             AnnouncementMessage::Refs(msg) => {
                 stmt.bind((2, &msg.rid))?;
                 stmt.bind((3, &GossipType::Refs))?;
                 stmt.bind((4, msg))?;
                 stmt.bind((4, &msg.encode_vec()[..]))?;
             }
             AnnouncementMessage::Inventory(msg) => {
                 stmt.bind((2, sql::Value::String(String::new())))?;
                 stmt.bind((3, &GossipType::Inventory))?;
                 stmt.bind((4, msg))?;
                 stmt.bind((4, &msg.encode_vec()[..]))?;
             }
         }
         stmt.bind((5, &ann.signature))?;

     }
 }
 impl sql::BindableWithIndex for &NodeAnnouncement {
     fn bind<I: sql::ParameterIndex>(self, stmt: &mut sql::Statement<'_>, i: I) -> sql::Result<()> {
         wire::serialize(self).bind(stmt, i)
     }
 }
 impl TryFrom<&sql::Value> for RefsAnnouncement {
     type Error = sql::Error;

     }
 }
 impl sql::BindableWithIndex for &InventoryAnnouncement {
     fn bind<I: sql::ParameterIndex>(self, stmt: &mut sql::Statement<'_>, i: I) -> sql::Result<()> {
         wire::serialize(self).bind(stmt, i)
     }
 }
 impl From<wire::Error> for sql::Error {
     fn from(other: wire::Error) -> Self {
         sql::Error {

 use crate::service::filter::Filter;
 use crate::service::{Link, NodeId, Timestamp};
 use crate::wire;
 use crate::wire::Encode as _;
 /// Maximum number of addresses which can be announced to other nodes.
 pub const ADDRESS_LIMIT: usize = 16;

         let mut output = [0u8; 32];
         scrypt::scrypt(
             wire::serialize(self).as_ref(),
             &self.encode_vec(),
             Announcement::POW_SALT,
             &params,
             &mut output,

     {
         use crypto::signature::Signer as _;
         let msg = wire::serialize(&self);
         let msg = self.encode_vec();
         let signature = signer.sign(&msg);
         Announcement {

     /// Verify this announcement's signature.
     pub fn verify(&self) -> bool {
         let msg = wire::serialize(&self.message);
         let msg = self.message.encode_vec();
         self.node.verify(msg, &self.signature).is_ok()
     }

 impl Ord for Message {
     fn cmp(&self, other: &Self) -> std::cmp::Ordering {
         let this = wire::serialize(self);
         let other = wire::serialize(other);
         let this = self.encode_vec();
         let other = other.encode_vec();
         this.cmp(&other)
     }

     use std::str::FromStr;
     use fastrand;
     use localtime::LocalTime;
     use qcheck_macros::quickcheck;
     use radicle::git::raw;
     use radicle::test::arbitrary;
     use crate::wire::Decode as _;
     use super::*;
     use crate::wire::Encode;
     use localtime::LocalTime;
     use radicle::test::arbitrary;
     #[test]
     fn test_ref_remote_limit() {

         let mut buf = Vec::new();
         msg.encode(&mut buf);
         let decoded = wire::deserialize(buf.as_slice());
         let decoded = Message::decode_exact(buf.as_slice());
         assert!(decoded.is_ok());
         assert_eq!(msg, decoded.unwrap());
     }

 use std::collections::BTreeMap;
 use std::convert::TryFrom;
 use std::fmt::Debug;
 use std::mem;
 use std::ops::Deref;
 use std::str::FromStr;

 pub type Size = u16;
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
 pub enum Invalid {
     #[error("invalid Git object identifier size: expected {expected}, got {actual}")]
     Oid { expected: usize, actual: usize },
     #[error(transparent)]
     Bounded(#[from] crate::bounded::Error),
     #[error("invalid filter size: {actual}")]
     FilterSize { actual: usize },
     #[error("UTF-8 error: {0}")]
     FromUtf8(#[from] FromUtf8Error),
     #[error("invalid size: expected {expected}, got {actual}")]
     InvalidSize { expected: usize, actual: usize },
     #[error("invalid filter size: {0}")]
     InvalidFilterSize(usize),
     #[error("invalid channel type {0:x}")]
     InvalidStreamKind(u8),
     #[error(transparent)]
     InvalidRefName(#[from] fmt::Error),
     RefName(#[from] fmt::Error),
     #[error(transparent)]
     InvalidAlias(#[from] node::AliasError),
     #[error("invalid user agent string: {0:?}")]
     InvalidUserAgent(String),
     #[error("invalid control message with type `{0}`")]
     InvalidControlMessage(u8),
     #[error("invalid protocol version header `{0:x?}`")]
     InvalidProtocolVersion([u8; 4]),
     Alias(#[from] node::AliasError),
     #[error("invalid user agent string: {err}")]
     InvalidUserAgent { err: String },
     #[error("invalid onion address: {0}")]
     InvalidOnionAddr(#[from] tor::OnionAddrDecodeError),
     #[error("invalid timestamp: {0}")]
     InvalidTimestamp(u64),
     #[error("wrong protocol version `{0}`")]
     WrongProtocolVersion(u8),
     #[error("unknown address type `{0}`")]
     UnknownAddressType(u8),
     #[error("unknown message type `{0}`")]
     UnknownMessageType(u16),
     #[error("unknown info type `{0}`")]
     UnknownInfoType(u16),
     #[error("unexpected bytes")]
     UnexpectedBytes,
     OnionAddr(#[from] tor::OnionAddrDecodeError),
     #[error("invalid timestamp: {actual_millis} millis")]
     Timestamp { actual_millis: u64 },
     // Message types
     #[error("invalid control message type: {actual:x}")]
     ControlType { actual: u8 },
     #[error("invalid stream type: {actual:x}")]
     StreamType { actual: u8 },
     #[error("invalid address type: {actual:x}")]
     AddressType { actual: u8 },
     #[error("invalid message type: {actual:x}")]
     MessageType { actual: u16 },
     #[error("invalid info message type: {actual:x}")]
     InfoMessageType { actual: u16 },
     // Protocol version handling
     #[error("invalid protocol version string: {actual:x?}")]
     ProtocolVersion { actual: [u8; 4] },
     #[error("unsupported protocol version: {actual}")]
     ProtocolVersionUnsupported { actual: u8 },
 }
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
     #[error(transparent)]
     Invalid(#[from] Invalid),
     #[error("unexpected end of buffer, requested {requested} more bytes but only {available} are available")]
     UnexpectedEnd { available: usize, requested: usize },
 }

 /// Things that can be encoded as binary.
 pub trait Encode {
     /// Encode self by writing it to the given buffer.
     fn encode(&self, buffer: &mut impl BufMut);
     /// A convenience wrapper around [`Encode::encode`]
     /// that allocates a [`Vec`].
     fn encode_vec(&self) -> Vec<u8> {
         let mut buf = Vec::new();
         self.encode(&mut buf);
         buf
     }
 }
 /// Things that can be decoded from binary.
 pub trait Decode: Sized {
     fn decode(buffer: &mut impl Buf) -> Result<Self, Error>;
 }
 /// Encode an object into a byte vector.
 ///
 /// # Panics
 ///
 /// If the encoded object exceeds [`Size::MAX`].
 pub fn serialize<E: Encode + ?Sized>(data: &E) -> Vec<u8> {
     let mut buffer = Vec::new().limit(Size::MAX as usize);
     data.encode(&mut buffer);
     buffer.into_inner()
 }
 /// Decode an object from a slice.
 pub fn deserialize<T: Decode>(mut data: &[u8]) -> Result<T, Error> {
     let result = T::decode(&mut data)?;
     if data.is_empty() {
         Ok(result)
     } else {
         Err(Error::UnexpectedBytes)
     /// A convenience wrapper around [`Decode::decode`] to decode
     /// from a slice exactly.
     ///
     /// # Panics
     ///
     ///  - If decoding failed because there were not enough bytes.
     ///  - If there are any bytes left after decoding.
     #[cfg(test)]
     fn decode_exact(mut data: &[u8]) -> Result<Self, Invalid> {
         match Self::decode(&mut data) {
             Ok(value) => {
                 if !data.is_empty() {
                     panic!("{} bytes left in buffer", data.len());
                 }
                 Ok(value)
             }
             Err(err @ Error::UnexpectedEnd { .. }) => {
                 panic!("{}", err);
             }
             Err(Error::Invalid(e)) => Err(e),
         }
     }
 }

         for _ in 0..len {
             let name = String::decode(buf)?;
             let name = git::RefString::try_from(name).map_err(Error::from)?;
             let name = git::RefString::try_from(name).map_err(Invalid::from)?;
             let oid = git::Oid::decode(buf)?;
             refs.insert(name, oid);

 impl Decode for git::RefString {
     fn decode(buf: &mut impl Buf) -> Result<Self, Error> {
         let ref_str = String::decode(buf)?;
         git::RefString::try_from(ref_str).map_err(Error::from)
         Ok(git::RefString::try_from(ref_str).map_err(Invalid::from)?)
     }
 }
 impl Decode for UserAgent {
     fn decode(buf: &mut impl Buf) -> Result<Self, Error> {
         String::decode(buf).and_then(|s| UserAgent::from_str(&s).map_err(Error::InvalidUserAgent))
         let user_agent = String::decode(buf)?;
         Ok(UserAgent::from_str(&user_agent).map_err(|err| Invalid::InvalidUserAgent { err })?)
     }
 }
 impl Decode for Alias {
     fn decode(buf: &mut impl Buf) -> Result<Self, Error> {
         String::decode(buf).and_then(|s| Alias::from_str(&s).map_err(Error::from))
         let alias = String::decode(buf)?;
         Ok(Alias::from_str(&alias).map_err(Invalid::from)?)
     }
 }

 impl Decode for git::Oid {
     fn decode(buf: &mut impl Buf) -> Result<Self, Error> {
         const LEN_EXPECTED: usize = mem::size_of::<git::raw::Oid>();
         let len = Size::decode(buf)? as usize;
         #[allow(non_upper_case_globals)]
         const expected: usize = mem::size_of::<git::raw::Oid>();
         if len != expected {
             return Err(Error::InvalidSize {
                 expected,
         if len != LEN_EXPECTED {
             return Err(Invalid::Oid {
                 expected: LEN_EXPECTED,
                 actual: len,
             });
             }
             .into());
         }
         let buf: [u8; expected] = Decode::decode(buf)?;
         let buf: [u8; LEN_EXPECTED] = Decode::decode(buf)?;
         let oid = git::raw::Oid::from_bytes(&buf).expect("the buffer is exactly the right size");
         let oid = git::Oid::from(oid);

 {
     fn decode(buf: &mut impl Buf) -> Result<Self, Error> {
         let len: usize = Size::decode(buf)? as usize;
         let mut items = Self::with_capacity(len).map_err(|_| Error::InvalidSize {
             expected: Self::max(),
             actual: len,
         })?;
         let mut items = Self::with_capacity(len).map_err(Invalid::from)?;
         for _ in 0..items.capacity() {
             let item = T::decode(buf)?;

         buf.try_copy_to_slice(&mut bytes)?;
         let string = String::from_utf8(bytes)?;
         let string = String::from_utf8(bytes).map_err(Invalid::from)?;
         Ok(string)
     }

     fn decode(buf: &mut impl Buf) -> Result<Self, Error> {
         let size: usize = Size::decode(buf)? as usize;
         if !filter::FILTER_SIZES.contains(&size) {
             return Err(Error::InvalidFilterSize(size));
             return Err(Invalid::FilterSize { actual: size }.into());
         }
         let mut bytes = vec![0; size];

 impl Decode for tor::OnionAddrV3 {
     fn decode(buf: &mut impl Buf) -> Result<Self, Error> {
         let bytes: [u8; tor::ONION_V3_RAW_LEN] = Decode::decode(buf)?;
         let addr = tor::OnionAddrV3::from_raw_bytes(bytes)?;
         let addr = tor::OnionAddrV3::from_raw_bytes(bytes).map_err(Invalid::from)?;
         Ok(addr)
     }

 impl Decode for Timestamp {
     fn decode(buf: &mut impl Buf) -> Result<Self, Error> {
         let millis = u64::decode(buf)?;
         let ts = Timestamp::try_from(millis).map_err(Error::InvalidTimestamp)?;
         let ts = Timestamp::try_from(millis).map_err(|value| Invalid::Timestamp {
             actual_millis: value,
         })?;
         Ok(ts)
     }

     #[quickcheck]
     fn prop_u8(input: u8) {
         assert_eq!(deserialize::<u8>(&serialize(&input)).unwrap(), input);
         assert_eq!(u8::decode_exact(&input.encode_vec()).unwrap(), input);
     }
     #[quickcheck]
     fn prop_u16(input: u16) {
         assert_eq!(deserialize::<u16>(&serialize(&input)).unwrap(), input);
         assert_eq!(u16::decode_exact(&input.encode_vec()).unwrap(), input);
     }
     #[quickcheck]
     fn prop_u32(input: u32) {
         assert_eq!(deserialize::<u32>(&serialize(&input)).unwrap(), input);
         assert_eq!(u32::decode_exact(&input.encode_vec()).unwrap(), input);
     }
     #[quickcheck]
     fn prop_u64(input: u64) {
         assert_eq!(deserialize::<u64>(&serialize(&input)).unwrap(), input);
         assert_eq!(u64::decode_exact(&input.encode_vec()).unwrap(), input);
     }
     #[quickcheck]

         if input.len() > u8::MAX as usize {
             return qcheck::TestResult::discard();
         }
         assert_eq!(deserialize::<String>(&serialize(&input)).unwrap(), input);
         assert_eq!(String::decode_exact(&input.encode_vec()).unwrap(), input);
         qcheck::TestResult::passed()
     }

     #[quickcheck]
     fn prop_vec(input: BoundedVec<String, 16>) {
         assert_eq!(
             deserialize::<BoundedVec<String, 16>>(&serialize(&input.as_slice())).unwrap(),
             BoundedVec::<String, 16>::decode_exact(&input.encode_vec()).unwrap(),
             input
         );
     }
     #[quickcheck]
     fn prop_pubkey(input: PublicKey) {
         assert_eq!(deserialize::<PublicKey>(&serialize(&input)).unwrap(), input);
         assert_eq!(PublicKey::decode_exact(&input.encode_vec()).unwrap(), input);
     }
     #[quickcheck]
     fn prop_filter(input: filter::Filter) {
         assert_eq!(
             deserialize::<filter::Filter>(&serialize(&input)).unwrap(),
             filter::Filter::decode_exact(&input.encode_vec()).unwrap(),
             input
         );
     }
     #[quickcheck]
     fn prop_id(input: RepoId) {
         assert_eq!(deserialize::<RepoId>(&serialize(&input)).unwrap(), input);
         assert_eq!(RepoId::decode_exact(&input.encode_vec()).unwrap(), input);
     }
     #[quickcheck]
     fn prop_refs(input: Refs) {
         assert_eq!(deserialize::<Refs>(&serialize(&input)).unwrap(), input);
         assert_eq!(Refs::decode_exact(&input.encode_vec()).unwrap(), input);
     }
     #[quickcheck]
     fn prop_tuple(input: (String, String)) {
         assert_eq!(
             deserialize::<(String, String)>(&serialize(&input)).unwrap(),
             <(String, String)>::decode_exact(&input.encode_vec()).unwrap(),
             input
         );
     }

         let signature = Signature::from(input);
         assert_eq!(
             deserialize::<Signature>(&serialize(&signature)).unwrap(),
             Signature::decode_exact(&signature.encode_vec()).unwrap(),
             signature
         );
     }

     fn prop_oid(input: [u8; 20]) {
         let oid = git::Oid::try_from(input.as_slice()).unwrap();
         assert_eq!(deserialize::<git::Oid>(&serialize(&oid)).unwrap(), oid);
         assert_eq!(git::Oid::decode_exact(&oid.encode_vec()).unwrap(), oid);
     }
     #[quickcheck]
     fn prop_signed_refs(input: SignedRefs<Unverified>) {
         assert_eq!(
             deserialize::<SignedRefs<Unverified>>(&serialize(&input)).unwrap(),
             SignedRefs::<Unverified>::decode_exact(&input.encode_vec()).unwrap(),
             input
         );
     }

     #[test]
     fn test_string() {
         assert_eq!(
             serialize(&String::from("hello")),
             String::from("hello").encode_vec(),
             vec![5, b'h', b'e', b'l', b'l', b'o']
         );
     }