Merge #94

94: No handwritten bytes r=TheNeikos a=matthiasbeyer



Co-authored-by: Matthias Beyer <mail@beyermatthias.de>

7 files changed, 161 insertions, 42 deletions

diff --git a/Cargo.lock b/Cargo.lock
index d4521ba..96f1ec9 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -600,6 +600,7 @@ dependencies = [
  "clap 3.2.23",
  "futures",
  "miette",
+ "mqtt-format",
  "textwrap 0.15.2",
  "tokio",
 ]
diff --git a/mqtt-format/src/v3/header.rs b/mqtt-format/src/v3/header.rs
index 3a02c18..f59ec1a 100644
--- a/mqtt-format/src/v3/header.rs
+++ b/mqtt-format/src/v3/header.rs
@@ -46,6 +46,38 @@ pub enum MPacketKind {
     Disconnect,
 }
 
+impl MPacketKind {
+    pub fn to_byte(&self) -> u8 {
+        match self {
+            MPacketKind::Connect => 1 << 4,
+            MPacketKind::Connack => 2 << 4,
+            MPacketKind::Publish { dup, qos, retain } => {
+                let upper = 0b0011;
+                let dup = if *dup { 0b1000 } else { 0 };
+                let retain = u8::from(*retain);
+                let qos = match qos {
+                    MQualityOfService::AtMostOnce => 0b0000,
+                    MQualityOfService::AtLeastOnce => 0b0001,
+                    MQualityOfService::ExactlyOnce => 0b0010,
+                } << 1;
+
+                (upper << 4) | dup | retain | qos
+            }
+            MPacketKind::Puback => 4 << 4,
+            MPacketKind::Pubrec => 5 << 4,
+            MPacketKind::Pubrel => (6 << 4) | 2,
+            MPacketKind::Pubcomp => 7 << 4,
+            MPacketKind::Subscribe => (8 << 4) | 2,
+            MPacketKind::Suback => 9 << 4,
+            MPacketKind::Unsubscribe => (10 << 4) | 2,
+            MPacketKind::Unsuback => 11 << 4,
+            MPacketKind::Pingreq => 12 << 4,
+            MPacketKind::Pingresp => 13 << 4,
+            MPacketKind::Disconnect => 14 << 4,
+        }
+    }
+}
+
 fn mpacketkind(input: &[u8]) -> IResult<&[u8], MPacketKind> {
     let (input, (upper, lower)): (_, (u8, u8)) =
         bits::<_, _, Error<(&[u8], usize)>, _, _>(tuple((
@@ -128,7 +160,7 @@ mod tests {
         qos::MQualityOfService,
     };
 
-    use super::{decode_variable_length, mfixedheader};
+    use super::{decode_variable_length, mfixedheader, mpacketkind};
 
     #[test]
     fn check_variable_length_decoding() {
@@ -170,4 +202,50 @@ mod tests {
 
         mfixedheader(input).unwrap_err();
     }
+
+    #[test]
+    fn check_roundtrip_packet_kind() {
+        fn test(mp: MPacketKind) {
+            assert_eq!(mp, mpacketkind(&[mp.to_byte()]).unwrap().1);
+        }
+
+        test(MPacketKind::Connect);
+        test(MPacketKind::Connack);
+        test(MPacketKind::Publish {
+            dup: false,
+            qos: MQualityOfService::AtMostOnce,
+            retain: false,
+        });
+        test(MPacketKind::Publish {
+            dup: false,
+            qos: MQualityOfService::AtLeastOnce,
+            retain: false,
+        });
+        test(MPacketKind::Publish {
+            dup: false,
+            qos: MQualityOfService::ExactlyOnce,
+            retain: false,
+        });
+        test(MPacketKind::Publish {
+            dup: true,
+            qos: MQualityOfService::AtMostOnce,
+            retain: false,
+        });
+        test(MPacketKind::Publish {
+            dup: false,
+            qos: MQualityOfService::AtMostOnce,
+            retain: true,
+        });
+        test(MPacketKind::Puback);
+        test(MPacketKind::Pubrec);
+        test(MPacketKind::Pubrel);
+        test(MPacketKind::Pubcomp);
+        test(MPacketKind::Subscribe);
+        test(MPacketKind::Suback);
+        test(MPacketKind::Unsubscribe);
+        test(MPacketKind::Unsuback);
+        test(MPacketKind::Pingreq);
+        test(MPacketKind::Pingresp);
+        test(MPacketKind::Disconnect);
+    }
 }
diff --git a/mqtt-tester/Cargo.toml b/mqtt-tester/Cargo.toml
index dd6ec06..b5c5a25 100644
--- a/mqtt-tester/Cargo.toml
+++ b/mqtt-tester/Cargo.toml
@@ -11,5 +11,6 @@ bytes = "1.1.0"
 clap = { version = "3.2.8", features = ["derive"] }
 futures = "0.3"
 miette = { version = "5.1.1", features = ["fancy"] }
+mqtt-format = { path = "../mqtt-format", version = "0.2.0" }
 textwrap = "0.15.0"
 tokio = { version = "1.20", features = ["macros", "process", "rt", "rt-multi-thread", "io-util", "time"] }
diff --git a/mqtt-tester/src/client_report.rs b/mqtt-tester/src/client_report.rs
index c020790..944ae1d 100644
--- a/mqtt-tester/src/client_report.rs
+++ b/mqtt-tester/src/client_report.rs
@@ -9,6 +9,13 @@ use std::process::Stdio;
 use std::time::Duration;
 
 use futures::FutureExt;
+use mqtt_format::v3::connect_return::MConnectReturnCode;
+use mqtt_format::v3::header::MPacketKind;
+use mqtt_format::v3::identifier::MPacketIdentifier;
+use mqtt_format::v3::packet::{MConnack, MPacket, MPublish, MSubscribe};
+use mqtt_format::v3::qos::MQualityOfService;
+use mqtt_format::v3::strings::MString;
+use mqtt_format::v3::subscription_request::MSubscriptionRequests;
 use tokio::process::Command;
 
 use crate::report::{Report, ReportResult};
@@ -48,24 +55,30 @@ async fn check_invalid_utf8_is_rejected(client_exe_path: &Path) -> miette::Resul
         .await
         .map(crate::command::Command::new)?
         .wait_for_write([
-            crate::command::ClientCommand::Send(&[
-                0b0010_0000, // CONNACK
-                0b0000_0010, // Remaining length
-                0b0000_0000, // No session present
-                0b0000_0000, // Connection accepted
-            ]),
-            crate::command::ClientCommand::Send(&[
-                0b0011_0000, // PUBLISH packet, DUP = 0, QoS = 0, Retain = 0
-                0b0000_0111, // Length
-                // Now the variable header
-                0b0000_0000,
-                0b0000_0010,
-                0x61,
-                0xC1,        // An invalid UTF-8 byte
-                0b0000_0000, // Packet identifier
-                0b0000_0001,
-                0x1, // Payload
-            ]),
+            crate::command::ClientCommand::Send(
+                crate::util::packet_to_vec(MPacket::Connack({
+                    MConnack {
+                        session_present: false,
+                        connect_return_code: MConnectReturnCode::Accepted,
+                    }
+                }))
+                .await?,
+            ),
+            crate::command::ClientCommand::Send(
+                [
+                    0b0011_0000, // PUBLISH packet, DUP = 0, QoS = 0, Retain = 0
+                    0b0000_0111, // Length
+                    // Now the variable header
+                    0b0000_0000,
+                    0b0000_0010,
+                    0x61,
+                    0xC1,        // An invalid UTF-8 byte
+                    0b0000_0000, // Packet identifier
+                    0b0000_0001,
+                    0x1, // Payload
+                ]
+                .to_vec(),
+            ),
         ]);
 
     let (result, output) = match tokio::time::timeout(Duration::from_millis(100), output).await {
@@ -95,26 +108,27 @@ async fn check_receiving_server_packet(client_exe_path: &Path) -> miette::Result
         .await
         .map(crate::command::Command::new)?
         .wait_for_write([
-            crate::command::ClientCommand::Send(&[
-                0b0010_0000, // CONNACK
-                0b0000_0010, // Remaining length
-                0b0000_0000, // No session present
-                0b0000_0000, // Connection accepted
-            ]),
-            crate::command::ClientCommand::Send(&[
-                0b1000_0010, // SUBSCRIBE packet
-                0b0000_1000, // Length
-                // Now the variable header
-                0b0000_0000, // Packet ID
-                0b0000_0001,
-                // First sub
-                0b0000_0000,
-                0b0000_0011, // Length
-                b'a',
-                b'/',
-                b'b',
-                0b0000_0001, // QoS
-            ]),
+            crate::command::ClientCommand::Send(
+                crate::util::packet_to_vec(MPacket::Connack({
+                    MConnack {
+                        session_present: false,
+                        connect_return_code: MConnectReturnCode::Accepted,
+                    }
+                }))
+                .await?,
+            ),
+            crate::command::ClientCommand::Send(
+                crate::util::packet_to_vec(MPacket::Subscribe({
+                    MSubscribe {
+                        id: MPacketIdentifier(1),
+                        subscriptions: MSubscriptionRequests {
+                            count: 1,
+                            data: b"a/b",
+                        },
+                    }
+                }))
+                .await?,
+            ),
         ]);
 
     let (result, output) = match tokio::time::timeout(Duration::from_millis(100), output).await {
diff --git a/mqtt-tester/src/command.rs b/mqtt-tester/src/command.rs
index f872928..e63c7f2 100644
--- a/mqtt-tester/src/command.rs
+++ b/mqtt-tester/src/command.rs
@@ -13,8 +13,8 @@ pub struct Command {
 }
 
 pub enum ClientCommand {
-    Send(&'static [u8]),
-    WaitFor(&'static [u8]),
+    Send(Vec<u8>),
+    WaitFor(Vec<u8>),
     WaitAndCheck(Box<dyn FnOnce(&[u8]) -> bool>),
 }
 
@@ -37,7 +37,9 @@ impl Command {
 
         for command in commands {
             match command {
-                ClientCommand::Send(bytes) => to_client.write_all(bytes).await.into_diagnostic()?,
+                ClientCommand::Send(bytes) => {
+                    to_client.write_all(&bytes).await.into_diagnostic()?
+                }
                 ClientCommand::WaitFor(expected_bytes) => {
                     let mut buf = Vec::with_capacity(expected_bytes.len());
                     match tokio::time::timeout(
diff --git a/mqtt-tester/src/main.rs b/mqtt-tester/src/main.rs
index aec2d08..9562931 100644
--- a/mqtt-tester/src/main.rs
+++ b/mqtt-tester/src/main.rs
@@ -7,6 +7,7 @@
 mod client_report;
 mod command;
 mod report;
+mod util;
 
 use std::{path::PathBuf, process::exit};
 
diff --git a/mqtt-tester/src/util.rs b/mqtt-tester/src/util.rs
new file mode 100644
index 0000000..b4e19ec
--- /dev/null
+++ b/mqtt-tester/src/util.rs
@@ -0,0 +1,22 @@
+//
+//   This Source Code Form is subject to the terms of the Mozilla Public
+//   License, v. 2.0. If a copy of the MPL was not distributed with this
+//   file, You can obtain one at http://mozilla.org/MPL/2.0/.
+//
+
+use std::pin::Pin;
+
+use miette::IntoDiagnostic;
+use mqtt_format::v3::packet::MPacket;
+
+pub async fn packet_to_vec<'m>(mpacket: MPacket<'m>) -> miette::Result<Vec<u8>> {
+    let mut buf = vec![];
+    {
+        let mut cursor = futures::io::Cursor::new(&mut buf);
+        mpacket
+            .write_to(Pin::new(&mut cursor))
+            .await
+            .into_diagnostic()?;
+    }
+    Ok(buf)
+}