flowsdk 0.4.2

Safety-first, realistic, behavior-predictable messaging SDK for MQTT and more.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

// SPDX-License-Identifier: MPL-2.0

use crate::mqtt_serde::control_packet::{ControlPacketType, MqttControlPacket, MqttPacket};
use crate::mqtt_serde::mqttv5::common::properties::{
    encode_properities_hdr, parse_properties_hdr, Property,
};
use crate::mqtt_serde::parser;
use crate::mqtt_serde::parser::{packet_type, parse_remaining_length, ParseError, ParseOk};
use serde::{Deserialize, Serialize};

#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
pub struct MqttConnAck {
    pub session_present: bool,
    pub reason_code: u8,
    pub properties: Option<Vec<Property>>,
}

impl MqttConnAck {
    pub fn new(session_present: bool, reason_code: u8, properties: Option<Vec<Property>>) -> Self {
        MqttConnAck {
            session_present,
            reason_code,
            properties,
        }
    }
}

impl MqttControlPacket for MqttConnAck {
    fn control_packet_type(&self) -> u8 {
        ControlPacketType::CONNACK as u8
    }

    // MQTT 5.0: 3.2.2, Variable header
    fn variable_header(&self) -> Result<Vec<u8>, parser::ParseError> {
        let mut bytes = Vec::new();
        // MQTT 5.0: 3.2.2.1, Session Present Flag
        bytes.push(if self.session_present { 1 } else { 0 });
        // MQTT 5.0: 3.2.2.2, Connect Reason Code
        bytes.push(self.reason_code);
        // MQTT 5.0: 3.2.2.3, CONNACK Properties
        let properties = self.properties.as_ref().cloned().unwrap_or_default();
        bytes.extend(encode_properities_hdr(&properties)?);
        Ok(bytes)
    }

    // MQTT 5.0: 3.2.3, CONNACK Payload
    fn payload(&self) -> Result<Vec<u8>, parser::ParseError> {
        // CONNACK has no payload
        Ok(Vec::new())
    }

    fn from_bytes(bytes: &[u8]) -> Result<ParseOk, ParseError> {
        parse_connack(bytes)
    }
}

pub fn parse_connack(buffer: &[u8]) -> Result<ParseOk, ParseError> {
    let packet_type = packet_type(buffer)?;
    if packet_type != ControlPacketType::CONNACK as u8 {
        return Err(ParseError::InvalidPacketType);
    }

    let (size, vbi_len) = parse_remaining_length(&buffer[1..])?;

    let mut offset = 1 + vbi_len;
    let total_len = offset + size;

    if total_len > buffer.len() {
        return Ok(ParseOk::Continue(total_len - buffer.len(), 0));
    }

    // MQTT 5.0 3.2.2.1 Connack Flag
    let session_present_byte = *buffer.get(offset).ok_or(ParseError::BufferTooShort)?;
    let session_present: bool = (session_present_byte & 0x01) == 1u8;

    // MQTT 5.0 Protocol Compliance: Reserved bits must be 0
    #[cfg(feature = "strict-protocol-compliance")]
    if session_present_byte & 0xFE != 0 {
        return Err(ParseError::ParseError(
            "CONNACK Connect Acknowledge Flags reserved bits must be 0".to_string(),
        ));
    }

    // MQTT 5.0: 3.2.2.2 Connect Reason Code
    let reason_code = *buffer.get(offset + 1).ok_or(ParseError::BufferTooShort)?;
    offset += 2; // move past the flags and reason code

    // MQTT 5.0: 3.2.2.3 Connack props
    let (properties, consumed) =
        parse_properties_hdr(buffer.get(offset..).ok_or(ParseError::BufferTooShort)?)?;
    offset += consumed;
    let properties = if properties.is_empty() {
        None
    } else {
        Some(properties)
    };

    // MQTT 5.0: 3.2.3 Connack Payload, NO payload
    if offset != total_len {
        return Err(ParseError::InvalidLength);
    }

    Ok(ParseOk::Packet(
        MqttPacket::ConnAck5(MqttConnAck::new(session_present, reason_code, properties)),
        offset,
    ))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::mqtt_serde::control_packet::MqttPacket;
    use crate::mqtt_serde::parser::ParseOk;

    #[test]
    fn test_parse_connack() {
        let packet_bytes: [u8; 24] = [
            0x20, 0x16, 0x00, 0x00, 0x13, 0x2a, 0x01, 0x29, 0x01, 0x28, 0x01, 0x27, 0x00, 0x10,
            0x00, 0x00, 0x25, 0x01, 0x22, 0xff, 0xff, 0x21, 0x00, 0x20,
        ];

        let result = MqttConnAck::from_bytes(&packet_bytes);
        match result {
            Ok(ParseOk::Packet(packet, consumed)) => {
                assert_eq!(consumed, 24);
                match packet {
                    MqttPacket::ConnAck5(connack) => {
                        assert!(!connack.session_present);
                        assert_eq!(connack.reason_code, 0);
                    }
                    _ => panic!("Invalid packet"),
                }
            }
            _ => panic!("Invalid result"),
        }
    }

    #[test]
    fn test_connack_invalid_reserved_flags() {
        // [MQTT-3.2.2-1]
        let connack = MqttConnAck::new(false, 0, None);
        let mut bytes = connack.to_bytes().unwrap();

        // A simple CONNACK has a remaining length of 2, so the var header starts at index 2.
        // The first byte of the var header is the Connect Acknowledge Flags.
        bytes[2] |= 0x02; // Set reserved bit 1 to 1

        let result = MqttConnAck::from_bytes(&bytes);
        assert!(matches!(result, Err(ParseError::ParseError(_))));
    }
}