wireforge_app/mqtt/
parser.rs

1//! MQTT packet parser.
2
3use std::str;
4
5use super::types::{MqttPacketType, MqttProperty, QoS};
6
7/// Decode MQTT variable-length remaining length (1-4 bytes).
8pub fn read_remaining_length(buf: &[u8]) -> Option<(u32, usize)> {
9    let mut value: u32 = 0;
10    let mut multiplier = 1u32;
11    for (i, &byte) in buf.iter().enumerate() {
12        value += (byte as u32 & 0x7F) * multiplier;
13        if byte & 0x80 == 0 { return Some((value, i + 1)); }
14        multiplier *= 128;
15        if multiplier > 128 * 128 * 128 { return None; }
16    }
17    None
18}
19
20pub fn write_remaining_length(len: u32) -> Vec<u8> {
21    let mut out = Vec::new();
22    let mut v = len;
23    loop {
24        let mut byte = (v % 128) as u8;
25        v /= 128;
26        if v > 0 { byte |= 0x80; }
27        out.push(byte);
28        if v == 0 { break; }
29    }
30    out
31}
32
33/// Parsed MQTT packet.
34#[derive(Debug, Clone)]
35pub struct MqttPacket<'a> {
36    buf: &'a [u8],
37    header_len: usize,
38    remaining_len: u32,
39}
40
41impl<'a> MqttPacket<'a> {
42    pub fn parse(buf: &'a [u8]) -> Option<Self> {
43        if buf.is_empty() { return None; }
44        let (remaining_len, header_len) = read_remaining_length(&buf[1..])?;
45        Some(Self { buf, header_len: 1 + header_len, remaining_len })
46    }
47
48    pub fn packet_type(&self) -> MqttPacketType { MqttPacketType::from(self.buf[0]) }
49    pub fn flags(&self) -> u8 { self.buf[0] & 0x0F }
50    pub fn qos(&self) -> QoS { QoS::from(self.flags()) }
51    pub fn remaining_length(&self) -> u32 { self.remaining_len }
52    pub fn variable_header(&self) -> &'a [u8] { &self.buf[self.header_len..] }
53    pub fn payload(&self) -> &'a [u8] {
54        let vh = self.variable_header();
55        let vh_len = self._vh_len();
56        &vh[vh_len..]
57    }
58
59    fn _vh_len(&self) -> usize {
60        match self.packet_type() {
61            MqttPacketType::Connect => {
62                let vh = self.variable_header();
63                if vh.len() < 6 { return 0; }
64                let plen = u16::from_be_bytes([vh[0], vh[1]]) as usize;
65                let mut pos = 2 + plen; // protocol name
66                if pos + 4 > vh.len() { return pos; }
67                pos += 1; // protocol level
68                let flags = vh[pos]; pos += 1; // connect flags
69                pos += 2; // keep_alive
70                // MQTT 5.0 properties
71                if vh[pos-4] == 5 { // protocol level 5
72                    if let Some((_, plen)) = read_remaining_length(&vh[pos..]) { pos += plen; }
73                }
74                // client_id
75                if pos + 2 > vh.len() { return pos; }
76                let cid_len = u16::from_be_bytes([vh[pos], vh[pos+1]]) as usize;
77                pos += 2 + cid_len;
78                // will
79                if flags & 0x04 != 0 && pos + 2 <= vh.len() {
80                    // MQTT 5.0: will properties
81                    if vh[2 + plen] == 5 {
82                        if let Some((_, plen)) = read_remaining_length(&vh[pos..]) { pos += plen; }
83                    }
84                    if pos + 2 <= vh.len() {
85                        let wt_len = u16::from_be_bytes([vh[pos], vh[pos+1]]) as usize;
86                        pos += 2 + wt_len;
87                        if pos + 2 <= vh.len() {
88                            let wm_len = u16::from_be_bytes([vh[pos], vh[pos+1]]) as usize;
89                            pos += 2 + wm_len;
90                        }
91                    }
92                }
93                // username
94                if flags & 0x80 != 0 && pos + 2 <= vh.len() {
95                    let ulen = u16::from_be_bytes([vh[pos], vh[pos+1]]) as usize;
96                    pos += 2 + ulen;
97                }
98                // password
99                if flags & 0x40 != 0 && pos + 2 <= vh.len() {
100                    let plen = u16::from_be_bytes([vh[pos], vh[pos+1]]) as usize;
101                    pos += 2 + plen;
102                }
103                pos
104            }
105            MqttPacketType::Connack => {
106                let vh = self.variable_header();
107                if vh.is_empty() { return 0; }
108                let mut pos = 2; // session_present + return_code
109                if vh.len() > 2 { // MQTT 5.0 properties
110                    if let Some((_, plen)) = read_remaining_length(&vh[pos..]) { pos += plen; }
111                }
112                pos
113            }
114            MqttPacketType::Publish => {
115                let vh = self.variable_header();
116                if vh.len() < 2 { return 0; }
117                let tlen = u16::from_be_bytes([vh[0], vh[1]]) as usize;
118                let mut pos = 2 + tlen;
119                if self.qos() as u8 > 0 { pos += 2; } // packet_id
120                pos
121            }
122            MqttPacketType::Subscribe => {
123                let vh = self.variable_header();
124                if vh.len() < 2 { return 0; }
125                2 // packet_id only
126            }
127            _ => 0,
128        }
129    }
130
131    pub fn message(&self) -> MqttMessage<'a> {
132        let vh = self.variable_header();
133        match self.packet_type() {
134            MqttPacketType::Connect => {
135                if vh.len() < 6 { return MqttMessage::Unknown { packet_type: MqttPacketType::Connect, data: vh }; }
136                let plen = u16::from_be_bytes([vh[0], vh[1]]) as usize;
137                let proto = str::from_utf8(&vh[2..2+plen]).unwrap_or("");
138                let level = vh[2+plen];
139                let flags = vh[3+plen];
140                let keep_alive = u16::from_be_bytes([vh[4+plen], vh[5+plen]]);
141                let mut pos = 6 + plen;
142                // MQTT 5 properties
143                let properties = if level == 5 {
144                    parse_properties(&vh[pos..]).unwrap_or_default()
145                } else { vec![] };
146                if level == 5 { if let Some((_, pl)) = read_remaining_length(&vh[pos..]) { pos += pl; } }
147                // client_id
148                let cid_len = if pos + 2 <= vh.len() { u16::from_be_bytes([vh[pos], vh[pos+1]]) as usize } else { 0 };
149                pos += 2;
150                let client_id = str::from_utf8(&vh[pos..pos+cid_len]).unwrap_or("");
151                pos += cid_len;
152                // will
153                let (will_topic, will_message) = if flags & 0x04 != 0 {
154                    let mut wt = None; let mut wm = None;
155                    if level == 5 { if let Some((_, pl)) = read_remaining_length(&vh[pos..]) { pos += pl; } }
156                    if pos + 2 <= vh.len() { let wtl = u16::from_be_bytes([vh[pos], vh[pos+1]]) as usize; pos += 2; if pos+wtl <= vh.len() { wt = str::from_utf8(&vh[pos..pos+wtl]).ok(); pos += wtl; } }
157                    if pos + 2 <= vh.len() { let wml = u16::from_be_bytes([vh[pos], vh[pos+1]]) as usize; pos += 2; if pos+wml <= vh.len() { wm = Some(&vh[pos..pos+wml]); pos += wml; } }
158                    (wt, wm)
159                } else { (None, None) };
160                // username/password
161                let username = if flags & 0x80 != 0 { if pos+2<=vh.len() { let ul = u16::from_be_bytes([vh[pos],vh[pos+1]]) as usize; pos+=2; str::from_utf8(&vh[pos..pos+ul]).ok() } else { None } } else { None };
162                let password = if flags & 0x40 != 0 { if pos+2<=vh.len() { let pl = u16::from_be_bytes([vh[pos],vh[pos+1]]) as usize; pos+=2; Some(&vh[pos..pos+pl]) } else { None } } else { None };
163                MqttMessage::Connect {
164                    protocol_name: proto, protocol_level: level,
165                    clean_start: flags & 0x02 != 0, keep_alive,
166                    client_id, will_topic, will_message, username, password, properties,
167                }
168            }
169            MqttPacketType::Publish => {
170                if vh.len() < 2 { return MqttMessage::Unknown { packet_type: MqttPacketType::Publish, data: vh }; }
171                let tlen = u16::from_be_bytes([vh[0], vh[1]]) as usize;
172                let topic = str::from_utf8(&vh[2..2+tlen]).unwrap_or("");
173                let mut pos = 2 + tlen;
174                let packet_id = if self.qos() as u8 > 0 && pos + 2 <= vh.len() { let id = u16::from_be_bytes([vh[pos], vh[pos+1]]); pos += 2; Some(id) } else { None };
175                let payload = &vh[pos..];
176                MqttMessage::Publish { topic, packet_id, payload }
177            }
178            MqttPacketType::Subscribe => {
179                let packet_id = if vh.len() >= 2 { u16::from_be_bytes([vh[0], vh[1]]) } else { 0 };
180                let topics = if vh.len() > 2 {
181                    let data = &vh[2..];
182                    let mut topics = Vec::new();
183                    let mut p = 0;
184                    while p + 3 <= data.len() {
185                        let tl = u16::from_be_bytes([data[p], data[p+1]]) as usize;
186                        p += 2;
187                        if p + tl < data.len() {
188                            let tf = str::from_utf8(&data[p..p+tl]).unwrap_or("");
189                            topics.push((tf, data[p+tl]));
190                            p += tl + 1;
191                        } else { break; }
192                    }
193                    topics
194                } else { vec![] };
195                MqttMessage::Subscribe { packet_id, topics }
196            }
197            _ => MqttMessage::Unknown { packet_type: self.packet_type(), data: vh },
198        }
199    }
200}
201
202fn parse_properties(data: &[u8]) -> Option<Vec<MqttProperty>> {
203    let (prop_len, mut pos) = read_remaining_length(data)?;
204    let end = (pos + prop_len as usize).min(data.len());
205    let mut props = Vec::new();
206    while pos < end {
207        let id = data[pos]; pos += 1;
208        match id {
209            0x11 => { if pos+4<=end { props.push(MqttProperty::SessionExpiryInterval(u32::from_be_bytes([data[pos],data[pos+1],data[pos+2],data[pos+3]]))); pos+=4; } }
210            0x27 => { if pos+4<=end { props.push(MqttProperty::MaximumPacketSize(u32::from_be_bytes([data[pos],data[pos+1],data[pos+2],data[pos+3]]))); pos+=4; } }
211            0x23 => { if pos+2<=end { props.push(MqttProperty::TopicAlias(u16::from_be_bytes([data[pos],data[pos+1]]))); pos+=2; } }
212            0x19 => { props.push(MqttProperty::RequestResponseInfo(data.get(pos).copied().unwrap_or(0) != 0)); pos += 1; }
213            _ => { let mut v = vec![id]; while pos < end { v.push(data[pos]); pos += 1; } props.push(MqttProperty::Unknown(id, v)); }
214        }
215    }
216    Some(props)
217}
218
219/// Parsed MQTT message (typed).
220#[derive(Debug, Clone)]
221pub enum MqttMessage<'a> {
222    Connect {
223        protocol_name: &'a str, protocol_level: u8, clean_start: bool, keep_alive: u16,
224        client_id: &'a str, will_topic: Option<&'a str>, will_message: Option<&'a [u8]>,
225        username: Option<&'a str>, password: Option<&'a [u8]>, properties: Vec<MqttProperty>,
226    },
227    Publish { topic: &'a str, packet_id: Option<u16>, payload: &'a [u8] },
228    Subscribe { packet_id: u16, topics: Vec<(&'a str, u8)> },
229    Unknown { packet_type: MqttPacketType, data: &'a [u8] },
230}
231
232#[cfg(test)]
233mod tests {
234    use super::*;
235
236    #[test]
237    fn remaining_length_encoding() {
238        assert_eq!(read_remaining_length(&[0x00]).unwrap(), (0, 1));
239        assert_eq!(read_remaining_length(&[0x7F]).unwrap(), (127, 1));
240        assert_eq!(read_remaining_length(&[0x80, 0x01]).unwrap(), (128, 2));
241        assert_eq!(read_remaining_length(&[0xFF, 0x7F]).unwrap(), (16383, 2));
242        assert_eq!(write_remaining_length(128), vec![0x80, 0x01]);
243        assert_eq!(write_remaining_length(127), vec![0x7F]);
244    }
245
246    #[test]
247    fn parse_mqtt311_connect() {
248        // MQTT 3.1.1 CONNECT: client_id="test"
249        let mut data = vec![0x10u8]; // type=Connect
250        // remaining length
251        let mut payload = vec![0x00, 0x04, b'M', b'Q', b'T', b'T']; // proto name
252        payload.push(4); // level=4 (3.1.1)
253        payload.push(0x02); // flags: clean_start
254        payload.extend_from_slice(&60u16.to_be_bytes()); // keep_alive
255        payload.extend_from_slice(&[0x00, 0x04, b't', b'e', b's', b't']); // client_id
256        data.extend_from_slice(&write_remaining_length(payload.len() as u32));
257        data.extend_from_slice(&payload);
258
259        let pkt = MqttPacket::parse(&data).unwrap();
260        assert_eq!(pkt.packet_type(), MqttPacketType::Connect);
261        match pkt.message() {
262            MqttMessage::Connect { client_id, clean_start, keep_alive, .. } => {
263                assert_eq!(client_id, "test"); assert!(clean_start); assert_eq!(keep_alive, 60);
264            }
265            _ => panic!("expected Connect"),
266        }
267    }
268
269    #[test]
270    fn parse_mqtt_publish() {
271        let topic = b"test/topic";
272        let payload = b"hello";
273        let mut vh = vec![];
274        vh.extend_from_slice(&(topic.len() as u16).to_be_bytes());
275        vh.extend_from_slice(topic);
276        // QoS 0, no packet_id
277        let mut data = vec![0x30u8]; // type=Publish, QoS=0
278        data.extend_from_slice(&write_remaining_length((vh.len() + payload.len()) as u32));
279        data.extend_from_slice(&vh);
280        data.extend_from_slice(payload);
281
282        let pkt = MqttPacket::parse(&data).unwrap();
283        match pkt.message() {
284            MqttMessage::Publish { topic: t, payload: p, .. } => {
285                assert_eq!(t, "test/topic"); assert_eq!(p, b"hello");
286            }
287            _ => panic!("expected Publish"),
288        }
289    }
290}
wireforge_app/mqtt/parser.rs

wireforge_app/mqtt/
parser.rs
