Support full done message

Currently, done messages are treated as having a zero-sized payload but
they are expected to have at least a 4 byte payload holding an error
code. This causes issues when serializing done messages that are
consumed applications expecting conformant done messages.

See https://kernel.org/doc/html/next/userspace-api/netlink/intro.html#netlink-message-types
for details.

Fixes #11

Author: ghananigans

src/done.rs

@@ -0,0 +1,134 @@
+// SPDX-License-Identifier: MIT
+
+use std::mem::size_of;
+
+use byteorder::{ByteOrder, NativeEndian};
+use netlink_packet_utils::DecodeError;
+
+use crate::{Emitable, Field, Parseable, Rest};
+
+const CODE: Field = 0..4;
+const EXTENDED_ACK: Rest = 4..;
+const DONE_HEADER_LEN: usize = EXTENDED_ACK.start;
+
+#[derive(Debug, PartialEq, Eq, Clone)]
+#[non_exhaustive]
+pub struct DoneBuffer<T> {
+    buffer: T,
+}
+
+impl<T: AsRef<[u8]>> DoneBuffer<T> {
+    pub fn new(buffer: T) -> DoneBuffer<T> {
+        DoneBuffer { buffer }
+    }
+
+    /// Consume the packet, returning the underlying buffer.
+    pub fn into_inner(self) -> T {
+        self.buffer
+    }
+
+    pub fn new_checked(buffer: T) -> Result<Self, DecodeError> {
+        let packet = Self::new(buffer);
+        packet.check_buffer_length()?;
+        Ok(packet)
+    }
+
+    fn check_buffer_length(&self) -> Result<(), DecodeError> {
+        let len = self.buffer.as_ref().len();
+        if len < DONE_HEADER_LEN {
+            Err(format!(
+                "invalid DoneBuffer: length is {len} but DoneBuffer are \
+                at least {DONE_HEADER_LEN} bytes"
+            )
+            .into())
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Return the error code
+    pub fn code(&self) -> i32 {
+        let data = self.buffer.as_ref();
+        NativeEndian::read_i32(&data[CODE])
+    }
+}
+
+impl<'a, T: AsRef<[u8]> + ?Sized> DoneBuffer<&'a T> {
+    /// Return a pointer to the extended ack attributes.
+    pub fn extended_ack(&self) -> &'a [u8] {
+        let data = self.buffer.as_ref();
+        &data[EXTENDED_ACK]
+    }
+}
+
+impl<'a, T: AsRef<[u8]> + AsMut<[u8]> + ?Sized> DoneBuffer<&'a mut T> {
+    /// Return a mutable pointer to the extended ack attributes.
+    pub fn extended_ack_mut(&mut self) -> &mut [u8] {
+        let data = self.buffer.as_mut();
+        &mut data[EXTENDED_ACK]
+    }
+}
+
+impl<T: AsRef<[u8]> + AsMut<[u8]>> DoneBuffer<T> {
+    /// set the error code field
+    pub fn set_code(&mut self, value: i32) {
+        let data = self.buffer.as_mut();
+        NativeEndian::write_i32(&mut data[CODE], value)
+    }
+}
+
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub struct DoneMessage {
+    pub code: i32,
+    pub extended_ack: Vec<u8>,
+}
+
+impl Emitable for DoneMessage {
+    fn buffer_len(&self) -> usize {
+        size_of::<i32>() + self.extended_ack.len()
+    }
+    fn emit(&self, buffer: &mut [u8]) {
+        let mut buffer = DoneBuffer::new(buffer);
+        buffer.set_code(self.code);
+        buffer
+            .extended_ack_mut()
+            .copy_from_slice(&self.extended_ack);
+    }
+}
+
+impl<'buffer, T: AsRef<[u8]> + 'buffer> Parseable<DoneBuffer<&'buffer T>>
+    for DoneMessage
+{
+    fn parse(buf: &DoneBuffer<&'buffer T>) -> Result<DoneMessage, DecodeError> {
+        Ok(DoneMessage {
+            code: buf.code(),
+            extended_ack: buf.extended_ack().to_vec(),
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn serialize_and_parse() {
+        let mut expected = DoneMessage::default();
+        expected.code = 5;
+        expected.extended_ack = vec![1, 2, 3];
+
+        let len = expected.buffer_len();
+        assert_eq!(len, size_of::<i32>() + expected.extended_ack.len());
+
+        let mut buf = vec![0; len];
+        expected.emit(&mut buf);
+
+        let done_buf = DoneBuffer::new(&buf);
+        assert_eq!(done_buf.code(), expected.code);
+        assert_eq!(done_buf.extended_ack(), &expected.extended_ack);
+
+        let got = DoneMessage::parse(&done_buf).unwrap();
+        assert_eq!(got, expected);
+    }
+}

src/lib.rs

@@ -244,6 +244,9 @@ pub(crate) type Field = Range<usize>;
 /// Represent a field that starts at a given index in a packet
 pub(crate) type Rest = RangeFrom<usize>;
 
+pub mod done;
+pub use self::done::*;
+
 pub mod error;
 pub use self::error::*;
 

src/message.rs

@@ -7,9 +7,9 @@ use netlink_packet_utils::DecodeError;
 
 use crate::{
     payload::{NLMSG_DONE, NLMSG_ERROR, NLMSG_NOOP, NLMSG_OVERRUN},
-    AckMessage, Emitable, ErrorBuffer, ErrorMessage, NetlinkBuffer,
-    NetlinkDeserializable, NetlinkHeader, NetlinkPayload, NetlinkSerializable,
-    Parseable,
+    AckMessage, DoneBuffer, DoneMessage, Emitable, ErrorBuffer, ErrorMessage,
+    NetlinkBuffer, NetlinkDeserializable, NetlinkHeader, NetlinkPayload,
+    NetlinkSerializable, Parseable,
 };
 
 /// Represent a netlink message.
@@ -98,9 +98,8 @@ where
         let bytes = buf.payload();
         let payload = match header.message_type {
             NLMSG_ERROR => {
-                let buf = ErrorBuffer::new_checked(&bytes)
-                    .context("failed to parse NLMSG_ERROR")?;
-                let msg = ErrorMessage::parse(&buf)
+                let msg = ErrorBuffer::new_checked(&bytes)
+                    .and_then(|buf| ErrorMessage::parse(&buf))
                     .context("failed to parse NLMSG_ERROR")?;
                 if msg.code >= 0 {
                     Ack(msg as AckMessage)
@@ -109,7 +108,12 @@ where
                 }
             }
             NLMSG_NOOP => Noop,
-            NLMSG_DONE => Done,
+            NLMSG_DONE => {
+                let msg = DoneBuffer::new_checked(&bytes)
+                    .and_then(|buf| DoneMessage::parse(&buf))
+                    .context("failed to parse NLMSG_DONE")?;
+                Done(msg)
+            }
             NLMSG_OVERRUN => Overrun(bytes.to_vec()),
             message_type => {
                 let inner_msg = I::deserialize(&header, bytes).context(
@@ -130,7 +134,8 @@ where
         use self::NetlinkPayload::*;
 
         let payload_len = match self.payload {
-            Noop | Done => 0,
+            Noop => 0,
+            Done(ref msg) => msg.buffer_len(),
             Overrun(ref bytes) => bytes.len(),
             Error(ref msg) => msg.buffer_len(),
             Ack(ref msg) => msg.buffer_len(),
@@ -148,7 +153,8 @@ where
         let buffer =
             &mut buffer[self.header.buffer_len()..self.header.length as usize];
         match self.payload {
-            Noop | Done => {}
+            Noop => {}
+            Done(ref msg) => msg.emit(buffer),
             Overrun(ref bytes) => buffer.copy_from_slice(bytes),
             Error(ref msg) => msg.emit(buffer),
             Ack(ref msg) => msg.emit(buffer),
@@ -168,3 +174,69 @@ where
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use std::{convert::Infallible, mem::size_of};
+
+    #[derive(Clone, Debug, Default, PartialEq)]
+    struct FakeNetlinkInnerMessage;
+
+    impl NetlinkSerializable for FakeNetlinkInnerMessage {
+        fn message_type(&self) -> u16 {
+            unimplemented!("unused by tests")
+        }
+
+        fn buffer_len(&self) -> usize {
+            unimplemented!("unused by tests")
+        }
+
+        fn serialize(&self, _buffer: &mut [u8]) {
+            unimplemented!("unused by tests")
+        }
+    }
+
+    impl NetlinkDeserializable for FakeNetlinkInnerMessage {
+        type Error = Infallible;
+
+        fn deserialize(
+            _header: &NetlinkHeader,
+            _payload: &[u8],
+        ) -> Result<Self, Self::Error> {
+            unimplemented!("unused by tests")
+        }
+    }
+
+    #[test]
+    fn test_done() {
+        let header = NetlinkHeader::default();
+        let mut done_msg = DoneMessage::default();
+        done_msg.code = 0;
+        done_msg.extended_ack = vec![6, 7, 8, 9];
+        let mut want = NetlinkMessage::new(
+            header,
+            NetlinkPayload::<FakeNetlinkInnerMessage>::Done(done_msg.clone()),
+        );
+        want.finalize();
+
+        let len = want.buffer_len();
+        assert_eq!(
+            len,
+            header.buffer_len()
+                + size_of::<i32>()
+                + done_msg.extended_ack.len()
+        );
+
+        let mut buf = vec![1; len];
+        want.emit(&mut buf);
+
+        let done_buf = DoneBuffer::new(&buf[header.buffer_len()..]);
+        assert_eq!(done_buf.code(), done_msg.code);
+        assert_eq!(done_buf.extended_ack(), &done_msg.extended_ack);
+
+        let got = NetlinkMessage::parse(&NetlinkBuffer::new(&buf)).unwrap();
+        assert_eq!(got, want);
+    }
+}

src/payload.rs

@@ -2,7 +2,7 @@
 
 use std::fmt::Debug;
 
-use crate::{AckMessage, ErrorMessage, NetlinkSerializable};
+use crate::{AckMessage, DoneMessage, ErrorMessage, NetlinkSerializable};
 
 /// The message is ignored.
 pub const NLMSG_NOOP: u16 = 1;
@@ -18,7 +18,7 @@ pub const NLMSG_ALIGNTO: u16 = 4;
 #[derive(Debug, PartialEq, Eq, Clone)]
 #[non_exhaustive]
 pub enum NetlinkPayload<I> {
-    Done,
+    Done(DoneMessage),
     Error(ErrorMessage),
     Ack(AckMessage),
     Noop,
@@ -32,7 +32,7 @@ where
 {
     pub fn message_type(&self) -> u16 {
         match self {
-            NetlinkPayload::Done => NLMSG_DONE,
+            NetlinkPayload::Done(_) => NLMSG_DONE,
             NetlinkPayload::Error(_) | NetlinkPayload::Ack(_) => NLMSG_ERROR,
             NetlinkPayload::Noop => NLMSG_NOOP,
             NetlinkPayload::Overrun(_) => NLMSG_OVERRUN,