node: Implement binary encoding for messages

Signed-off-by: Alexis Sellier <self@cloudhead.io>
2026-11-12 14:40:37 +08:00 · 2022-09-13 10:16:05 +02:00
parent c7f78eafb6
commit b3d3077ccc
11 changed files with 638 additions and 88 deletions
@@ -1,10 +1,12 @@
 use std::net;

-use git_url::Url;
+use byteorder::NetworkEndian;
 use serde::{Deserialize, Serialize};

 use crate::crypto;
+use crate::git;
 use crate::identity::Id;
+use crate::protocol::wire;
 use crate::protocol::{Context, NodeId, Timestamp, PROTOCOL_VERSION};
 use crate::storage;
 use crate::storage::refs::SignedRefs;
@@ -28,13 +30,6 @@ pub struct Hostname(String);
 /// Peer public protocol address.
 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
 pub enum Address {
-    /// Tor V3 onion address.
-    Onion {
-        key: crypto::PublicKey,
-        port: u16,
-        checksum: u16,
-        version: u8,
-    },
    Ip {
        ip: net::IpAddr,
        port: u16,
@@ -43,6 +38,57 @@ pub enum Address {
        host: Hostname,
        port: u16,
    },
+    /// Tor V3 onion address.
+    Onion {
+        key: crypto::PublicKey,
+        port: u16,
+        checksum: u16,
+        version: u8,
+    },
+}
+
+impl wire::Encode for Address {
+    fn encode<W: std::io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, std::io::Error> {
+        let mut n = 0;
+
+        match self {
+            Self::Ip { ip, port } => {
+                match ip {
+                    net::IpAddr::V4(addr) => {
+                        n += 1u8.encode(writer)?;
+                        n += addr.octets().encode(writer)?;
+                    }
+                    net::IpAddr::V6(addr) => {
+                        n += 2u8.encode(writer)?;
+                        n += addr.octets().encode(writer)?;
+                    }
+                }
+                n += port.encode(writer)?;
+            }
+            Self::Hostname { .. } => todo!(),
+            Self::Onion { .. } => todo!(),
+        }
+        Ok(n)
+    }
+}
+
+impl wire::Decode for Address {
+    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
+        use byteorder::ReadBytesExt;
+
+        match reader.read_u8()? {
+            1 => {
+                let octets: [u8; 4] = wire::Decode::decode(reader)?;
+                let ip = net::IpAddr::from(net::Ipv4Addr::from(octets));
+                let port = u16::decode(reader)?;
+
+                Ok(Self::Ip { ip, port })
+            }
+            _ => {
+                todo!();
+            }
+        }
+    }
 }

 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
@@ -78,7 +124,7 @@ pub enum Message {
        timestamp: Timestamp,
        version: u32,
        addrs: Vec<Address>,
-        git: Url,
+        git: git::Url,
    },
    Node {
        /// Signature over the announcement, by the node being announced.
@@ -91,11 +137,9 @@ pub enum Message {
    /// Send our inventory to a peer. Sent in response to [`Message::GetInventory`].
    /// Nb. This should be the whole inventory, not a partial update.
    Inventory {
+        node: NodeId,
        inv: Vec<Id>,
        timestamp: Timestamp,
-        /// Original peer this inventory came from. We don't set this when we
-        /// are the originator, only when relaying.
-        origin: Option<NodeId>,
    },
    /// Project refs were updated. Includes the signature of the user who updated
    /// their view of the project.
@@ -110,7 +154,7 @@ pub enum Message {
 }

 impl Message {
-    pub fn hello(id: NodeId, timestamp: Timestamp, addrs: Vec<Address>, git: Url) -> Self {
+    pub fn hello(id: NodeId, timestamp: Timestamp, addrs: Vec<Address>, git: git::Url) -> Self {
        Self::Hello {
            id,
            timestamp,
@@ -133,18 +177,127 @@ impl Message {
    pub fn inventory<S, T, G>(ctx: &Context<S, T, G>) -> Result<Self, storage::Error>
    where
        T: storage::ReadStorage,
+        G: crypto::Signer,
    {
        let timestamp = ctx.timestamp();
        let inv = ctx.storage.inventory()?;

        Ok(Self::Inventory {
-            timestamp,
+            node: ctx.id(),
            inv,
-            origin: None,
+            timestamp,
        })
    }

    pub fn get_inventory(ids: impl Into<Vec<Id>>) -> Self {
        Self::GetInventory { ids: ids.into() }
    }
+
+    pub fn type_id(&self) -> u16 {
+        match self {
+            Self::Hello { .. } => 0,
+            Self::Node { .. } => 2,
+            Self::GetInventory { .. } => 4,
+            Self::Inventory { .. } => 6,
+            Self::RefsUpdate { .. } => 8,
+        }
+    }
+}
+
+impl wire::Encode for Message {
+    fn encode<W: std::io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, std::io::Error> {
+        let mut n = self.type_id().encode(writer)?;
+
+        match self {
+            Self::Hello {
+                id,
+                timestamp,
+                version,
+                addrs,
+                git,
+            } => {
+                n += id.encode(writer)?;
+                n += timestamp.encode(writer)?;
+                n += version.encode(writer)?;
+                n += addrs.as_slice().encode(writer)?;
+                n += git.encode(writer)?;
+            }
+            Self::RefsUpdate { id, signer, refs } => {
+                n += id.encode(writer)?;
+                n += signer.encode(writer)?;
+                n += refs.encode(writer)?;
+            }
+            Self::GetInventory { ids } => {
+                n += ids.as_slice().encode(writer)?;
+            }
+            Self::Inventory {
+                node,
+                inv,
+                timestamp,
+            } => {
+                n += node.encode(writer)?;
+                n += inv.as_slice().encode(writer)?;
+                n += timestamp.encode(writer)?;
+            }
+            Self::Node { .. } => {
+                todo!();
+            }
+        }
+        Ok(n)
+    }
+}
+
+impl wire::Decode for Message {
+    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
+        use byteorder::ReadBytesExt;
+
+        let type_id = reader.read_u16::<NetworkEndian>()?;
+
+        match type_id {
+            0 => {
+                let id = NodeId::decode(reader)?;
+                let timestamp = Timestamp::decode(reader)?;
+                let version = u32::decode(reader)?;
+                let addrs = Vec::<Address>::decode(reader)?;
+                let git = git::Url::decode(reader)?;
+
+                Ok(Self::Hello {
+                    id,
+                    timestamp,
+                    version,
+                    addrs,
+                    git,
+                })
+            }
+            2 => {
+                todo!();
+            }
+            4 => {
+                let ids = Vec::<Id>::decode(reader)?;
+
+                Ok(Self::GetInventory { ids })
+            }
+            6 => {
+                let node = NodeId::decode(reader)?;
+                let inv = Vec::<Id>::decode(reader)?;
+                let timestamp = Timestamp::decode(reader)?;
+
+                Ok(Self::Inventory {
+                    node,
+                    inv,
+                    timestamp,
+                })
+            }
+            8 => {
+                let id = Id::decode(reader)?;
+                let signer = crypto::PublicKey::decode(reader)?;
+                let refs = SignedRefs::decode(reader)?;
+
+                Ok(Self::RefsUpdate { id, signer, refs })
+            }
+            _ => {
+                todo!();
+            }
+        }
+    }
 }
@@ -163,11 +163,11 @@ impl Peer {
                ctx.write(self.addr, inventory);
            }
            (
-                PeerState::Negotiated { id, git, .. },
+                PeerState::Negotiated { git, .. },
                Message::Inventory {
+                    node,
                    timestamp,
                    inv,
-                    origin,
                },
            ) => {
                let now = ctx.clock.local_time();
@@ -184,13 +184,13 @@ impl Peer {
                } else {
                    return Ok(None);
                }
-                ctx.process_inventory(&inv, origin.unwrap_or(*id), git);
+                ctx.process_inventory(&inv, node, git);

                if ctx.config.relay {
                    return Ok(Some(Message::Inventory {
-                        timestamp,
+                        node,
                        inv,
-                        origin: origin.or(Some(*id)),
+                        timestamp,
                    }));
                }
            }
@@ -0,0 +1,348 @@
+use std::collections::BTreeMap;
+use std::convert::TryFrom;
+use std::ops::Deref;
+use std::string::FromUtf8Error;
+use std::{io, mem, net};
+
+use byteorder::{NetworkEndian, ReadBytesExt, WriteBytesExt};
+
+use crate::crypto::{PublicKey, Signature};
+use crate::git;
+use crate::git::fmt;
+use crate::hash::Digest;
+use crate::identity::Id;
+use crate::storage::refs::Refs;
+
+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+    #[error("i/o: {0}")]
+    Io(#[from] io::Error),
+    #[error("UTF-8 error: {0}")]
+    FromUtf8(#[from] FromUtf8Error),
+    #[error("invalid size: expected {expected}, got {actual}")]
+    InvalidSize { expected: usize, actual: usize },
+    #[error(transparent)]
+    InvalidRefName(#[from] fmt::Error),
+}
+
+pub trait Encode {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error>;
+}
+
+pub trait Decode: Sized {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error>;
+}
+
+/// Encode an object into a vector.
+pub fn serialize<T: Encode + ?Sized>(data: &T) -> Vec<u8> {
+    let mut buffer = Vec::new();
+    let len = data
+        .encode(&mut buffer)
+        .expect("in-memory writes don't error");
+
+    debug_assert_eq!(len, buffer.len());
+
+    buffer
+}
+
+impl Encode for u8 {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        writer.write_u8(*self)?;
+
+        Ok(mem::size_of::<Self>())
+    }
+}
+
+impl Encode for u16 {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        writer.write_u16::<NetworkEndian>(*self)?;
+
+        Ok(mem::size_of::<Self>())
+    }
+}
+
+impl Encode for u32 {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        writer.write_u32::<NetworkEndian>(*self)?;
+
+        Ok(mem::size_of::<Self>())
+    }
+}
+
+impl Encode for u64 {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        writer.write_u64::<NetworkEndian>(*self)?;
+
+        Ok(mem::size_of::<Self>())
+    }
+}
+
+impl Encode for usize {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        let u = (*self)
+            .try_into()
+            .map_err(|_| io::Error::from(io::ErrorKind::InvalidInput))?;
+
+        writer.write_u32::<NetworkEndian>(u)?;
+
+        Ok(mem::size_of_val(&u))
+    }
+}
+
+impl Encode for PublicKey {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        self.as_bytes().encode(writer)
+    }
+}
+
+impl<const T: usize> Encode for &[u8; T] {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        writer.write_all(*self)?;
+
+        Ok(mem::size_of::<Self>())
+    }
+}
+
+impl<const T: usize> Encode for [u8; T] {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        writer.write_all(self)?;
+
+        Ok(mem::size_of::<Self>())
+    }
+}
+
+impl<T> Encode for &[T]
+where
+    T: Encode,
+{
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        let mut n = self.len().encode(writer)?;
+
+        for item in self.iter() {
+            n += item.encode(writer)?;
+        }
+        Ok(n)
+    }
+}
+
+impl Encode for net::IpAddr {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        match self {
+            net::IpAddr::V4(addr) => addr.octets().encode(writer),
+            net::IpAddr::V6(addr) => addr.octets().encode(writer),
+        }
+    }
+}
+
+impl Encode for str {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        let mut n = 0;
+
+        n += self.len().encode(writer)?;
+        n += self.as_bytes().encode(writer)?;
+
+        Ok(n)
+    }
+}
+
+impl Encode for String {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        self.as_str().encode(writer)
+    }
+}
+
+impl Encode for git::Url {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        self.to_string().encode(writer)
+    }
+}
+
+impl Encode for Digest {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        self.as_ref().encode(writer)
+    }
+}
+
+impl Encode for Id {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        self.deref().encode(writer)
+    }
+}
+
+impl Encode for Refs {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        let mut n = self.len().encode(writer)?;
+
+        for (name, oid) in self.iter() {
+            n += name.as_str().encode(writer)?;
+            n += oid.encode(writer)?;
+        }
+        Ok(n)
+    }
+}
+
+impl Encode for Signature {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        self.to_bytes().encode(writer)
+    }
+}
+
+impl Encode for git::Oid {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        // Nb. We use length-encoding here to support future SHA-2 object ids.
+        self.as_bytes().encode(writer)
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+impl Decode for PublicKey {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        let buf: [u8; 32] = Decode::decode(reader)?;
+
+        PublicKey::try_from(buf)
+            .map_err(|e| Error::Io(io::Error::new(io::ErrorKind::InvalidInput, e.to_string())))
+    }
+}
+
+impl Decode for Refs {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        let len = usize::decode(reader)?;
+        let mut refs = BTreeMap::new();
+
+        for _ in 0..len {
+            let name = String::decode(reader)?;
+            let name = git::RefString::try_from(name).map_err(Error::from)?;
+            let oid = git::Oid::decode(reader)?;
+
+            refs.insert(name, oid);
+        }
+        Ok(refs.into())
+    }
+}
+
+impl Decode for git::Oid {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        let len = usize::decode(reader)?;
+        #[allow(non_upper_case_globals)]
+        const expected: usize = mem::size_of::<git2::Oid>();
+
+        if len != expected {
+            return Err(Error::InvalidSize {
+                expected,
+                actual: len,
+            });
+        }
+
+        let buf: [u8; expected] = Decode::decode(reader)?;
+        let oid = git2::Oid::from_bytes(&buf).expect("the buffer is exactly the right size");
+        let oid = git::Oid::from(oid);
+
+        Ok(oid)
+    }
+}
+
+impl Decode for Signature {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        let bytes: [u8; 64] = Decode::decode(reader)?;
+
+        Ok(Signature::from(bytes))
+    }
+}
+
+impl Decode for u8 {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        reader.read_u8().map_err(Error::from)
+    }
+}
+
+impl Decode for u16 {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        reader.read_u16::<NetworkEndian>().map_err(Error::from)
+    }
+}
+
+impl Decode for u32 {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        reader.read_u32::<NetworkEndian>().map_err(Error::from)
+    }
+}
+
+impl Decode for u64 {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        reader.read_u64::<NetworkEndian>().map_err(Error::from)
+    }
+}
+
+impl Decode for usize {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        let size: usize = u64::decode(reader)?
+            .try_into()
+            .map_err(|_| io::Error::from(io::ErrorKind::InvalidInput))?;
+
+        Ok(size)
+    }
+}
+
+impl<const N: usize> Decode for [u8; N] {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        let mut ary = [0; N];
+        reader.read_exact(&mut ary)?;
+
+        Ok(ary)
+    }
+}
+
+impl<T> Decode for Vec<T>
+where
+    T: Decode,
+{
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        let len: usize = usize::decode(reader)?;
+        let mut vec = Vec::with_capacity(len);
+
+        for _ in 0..len {
+            let item = T::decode(reader)?;
+            vec.push(item);
+        }
+        Ok(vec)
+    }
+}
+
+impl Decode for String {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        let len = usize::decode(reader)?;
+        let mut bytes = vec![0; len];
+
+        reader.read_exact(&mut bytes)?;
+
+        let string = String::from_utf8(bytes)?;
+
+        Ok(string)
+    }
+}
+
+impl Decode for git::Url {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        let string = String::decode(reader)?;
+
+        Self::from_bytes(string.as_bytes())
+            .map_err(|e| Error::Io(io::Error::new(io::ErrorKind::InvalidInput, e.to_string())))
+    }
+}
+
+impl Decode for Id {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        let digest: Digest = Decode::decode(reader)?;
+
+        Ok(Self::from(digest))
+    }
+}
+
+impl Decode for Digest {
+    fn decode<R: io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        let bytes: [u8; 32] = Decode::decode(reader)?;
+
+        Ok(Self::from(bytes))
+    }
+}