netget/server/mqtt_sn/
packet.rs

1//! MQTT-SN v1.2 packets: the 1- or 3-byte length header, every message type, flags, the three
2//! topic id types and forwarder encapsulation (section 5.5).
3use anyhow::{bail, ensure, Context, Result};
4
5pub const PROTOCOL_ID: u8 = 0x01;
6/// Return codes (section 5.3.10).
7pub const ACCEPTED: u8 = 0x00;
8pub const CONGESTION: u8 = 0x01;
9pub const INVALID_TOPIC_ID: u8 = 0x02;
10pub const NOT_SUPPORTED: u8 = 0x03;
11
12pub fn return_code_name(rc: u8) -> &'static str {
13    match rc {
14        ACCEPTED => "accepted",
15        CONGESTION => "congestion",
16        INVALID_TOPIC_ID => "invalid_topic_id",
17        NOT_SUPPORTED => "not_supported",
18        _ => "unknown",
19    }
20}
21
22pub fn return_code(name: &str) -> Option<u8> {
23    [CONGESTION, INVALID_TOPIC_ID, NOT_SUPPORTED]
24        .into_iter()
25        .find(|rc| return_code_name(*rc) == name)
26}
27
28/// QoS as carried in the flags: 0, 1, 2 or -1 (publish without a connection).
29pub type Qos = i8;
30
31#[derive(Clone, Copy, Debug, PartialEq, Default)]
32pub struct Flags {
33    pub dup: bool,
34    pub qos: Qos,
35    pub retain: bool,
36    pub will: bool,
37    pub clean_session: bool,
38    /// 0 normal topic id or name, 1 predefined id, 2 short name.
39    pub topic_id_type: u8,
40}
41
42impl Flags {
43    pub fn decode(b: u8) -> Self {
44        Self {
45            dup: b & 0x80 != 0,
46            qos: match (b >> 5) & 0x03 {
47                0 => 0,
48                1 => 1,
49                2 => 2,
50                _ => -1,
51            },
52            retain: b & 0x10 != 0,
53            will: b & 0x08 != 0,
54            clean_session: b & 0x04 != 0,
55            topic_id_type: b & 0x03,
56        }
57    }
58    pub fn encode(&self) -> u8 {
59        let q = match self.qos {
60            0 => 0,
61            1 => 1,
62            2 => 2,
63            _ => 3,
64        };
65        (u8::from(self.dup) << 7)
66            | (q << 5)
67            | (u8::from(self.retain) << 4)
68            | (u8::from(self.will) << 3)
69            | (u8::from(self.clean_session) << 2)
70            | (self.topic_id_type & 0x03)
71    }
72}
73
74pub const TOPIC_NORMAL: u8 = 0;
75pub const TOPIC_PREDEFINED: u8 = 1;
76pub const TOPIC_SHORT: u8 = 2;
77
78#[derive(Clone, Debug, PartialEq)]
79pub enum Packet {
80    Advertise {
81        gw_id: u8,
82        duration: u16,
83    },
84    SearchGw {
85        radius: u8,
86    },
87    GwInfo {
88        gw_id: u8,
89        gw_add: Vec<u8>,
90    },
91    Connect {
92        flags: Flags,
93        duration: u16,
94        client_id: String,
95    },
96    ConnAck {
97        rc: u8,
98    },
99    WillTopicReq,
100    WillTopic {
101        flags: Flags,
102        topic: String,
103    },
104    WillMsgReq,
105    WillMsg {
106        msg: Vec<u8>,
107    },
108    Register {
109        topic_id: u16,
110        msg_id: u16,
111        topic: String,
112    },
113    RegAck {
114        topic_id: u16,
115        msg_id: u16,
116        rc: u8,
117    },
118    /// `topic` is the topic id, the predefined id, or the two short-name bytes, per the flags.
119    Publish {
120        flags: Flags,
121        topic: u16,
122        msg_id: u16,
123        data: Vec<u8>,
124    },
125    PubAck {
126        topic_id: u16,
127        msg_id: u16,
128        rc: u8,
129    },
130    PubComp {
131        msg_id: u16,
132    },
133    PubRec {
134        msg_id: u16,
135    },
136    PubRel {
137        msg_id: u16,
138    },
139    /// A topic name (normal), or an id / short name in `topic_id`.
140    Subscribe {
141        flags: Flags,
142        msg_id: u16,
143        topic_name: Option<String>,
144        topic_id: u16,
145    },
146    SubAck {
147        flags: Flags,
148        topic_id: u16,
149        msg_id: u16,
150        rc: u8,
151    },
152    Unsubscribe {
153        flags: Flags,
154        msg_id: u16,
155        topic_name: Option<String>,
156        topic_id: u16,
157    },
158    UnsubAck {
159        msg_id: u16,
160    },
161    PingReq {
162        client_id: Option<String>,
163    },
164    PingResp,
165    Disconnect {
166        duration: Option<u16>,
167    },
168    WillTopicUpd {
169        flags: Flags,
170        topic: String,
171    },
172    WillTopicResp {
173        rc: u8,
174    },
175    WillMsgUpd {
176        msg: Vec<u8>,
177    },
178    WillMsgResp {
179        rc: u8,
180    },
181}
182
183impl Packet {
184    pub fn name(&self) -> &'static str {
185        match self {
186            Packet::Advertise { .. } => "ADVERTISE",
187            Packet::SearchGw { .. } => "SEARCHGW",
188            Packet::GwInfo { .. } => "GWINFO",
189            Packet::Connect { .. } => "CONNECT",
190            Packet::ConnAck { .. } => "CONNACK",
191            Packet::WillTopicReq => "WILLTOPICREQ",
192            Packet::WillTopic { .. } => "WILLTOPIC",
193            Packet::WillMsgReq => "WILLMSGREQ",
194            Packet::WillMsg { .. } => "WILLMSG",
195            Packet::Register { .. } => "REGISTER",
196            Packet::RegAck { .. } => "REGACK",
197            Packet::Publish { .. } => "PUBLISH",
198            Packet::PubAck { .. } => "PUBACK",
199            Packet::PubComp { .. } => "PUBCOMP",
200            Packet::PubRec { .. } => "PUBREC",
201            Packet::PubRel { .. } => "PUBREL",
202            Packet::Subscribe { .. } => "SUBSCRIBE",
203            Packet::SubAck { .. } => "SUBACK",
204            Packet::Unsubscribe { .. } => "UNSUBSCRIBE",
205            Packet::UnsubAck { .. } => "UNSUBACK",
206            Packet::PingReq { .. } => "PINGREQ",
207            Packet::PingResp => "PINGRESP",
208            Packet::Disconnect { .. } => "DISCONNECT",
209            Packet::WillTopicUpd { .. } => "WILLTOPICUPD",
210            Packet::WillTopicResp { .. } => "WILLTOPICRESP",
211            Packet::WillMsgUpd { .. } => "WILLMSGUPD",
212            Packet::WillMsgResp { .. } => "WILLMSGRESP",
213        }
214    }
215}
216
217fn u16_at(b: &[u8], i: usize) -> Result<u16> {
218    ensure!(b.len() >= i + 2, "the message is truncated");
219    Ok(u16::from_be_bytes([b[i], b[i + 1]]))
220}
221
222fn text(b: &[u8]) -> Result<String> {
223    String::from_utf8(b.to_vec()).context("a topic or client id is not UTF-8")
224}
225
226/// The message at the front of a datagram: (body after the header, message type, total length).
227fn header(d: &[u8]) -> Result<(usize, u8, usize)> {
228    ensure!(d.len() >= 2, "the datagram is shorter than a header");
229    let (len, at) = if d[0] == 0x01 {
230        ensure!(d.len() >= 4, "the 3-byte length is truncated");
231        (u16::from_be_bytes([d[1], d[2]]) as usize, 3)
232    } else {
233        (d[0] as usize, 1)
234    };
235    ensure!(
236        len > at && len <= d.len(),
237        "the length field {len} does not fit the datagram"
238    );
239    Ok((at + 1, d[at], len))
240}
241
242/// A forwarder-encapsulated message's wireless node id and the message inside.
243pub fn unwrap_forwarder(d: &[u8]) -> Result<Option<(Vec<u8>, &[u8])>> {
244    let (body, kind, len) = header(d)?;
245    if kind != 0xFE {
246        return Ok(None);
247    }
248    ensure!(len > body, "the encapsulation has no control byte");
249    Ok(Some((d[body + 1..len].to_vec(), &d[len..])))
250}
251
252pub fn wrap_forwarder(node: &[u8], inner: &[u8]) -> Vec<u8> {
253    let mut out = vec![(3 + node.len()) as u8, 0xFE, 0x00];
254    out.extend(node);
255    out.extend(inner);
256    out
257}
258
259pub fn decode(d: &[u8]) -> Result<Packet> {
260    let (at, kind, len) = header(d)?;
261    ensure!(
262        len == d.len(),
263        "the datagram carries {} bytes after the message",
264        d.len() - len
265    );
266    let b = &d[at..len];
267    let need = |n: usize| -> Result<()> {
268        ensure!(b.len() >= n, "the {kind:#04x} message is truncated");
269        Ok(())
270    };
271    Ok(match kind {
272        0x00 => {
273            need(3)?;
274            Packet::Advertise {
275                gw_id: b[0],
276                duration: u16_at(b, 1)?,
277            }
278        }
279        0x01 => {
280            need(1)?;
281            Packet::SearchGw { radius: b[0] }
282        }
283        0x02 => {
284            need(1)?;
285            Packet::GwInfo {
286                gw_id: b[0],
287                gw_add: b[1..].to_vec(),
288            }
289        }
290        0x04 => {
291            need(4)?;
292            ensure!(
293                b[1] == PROTOCOL_ID,
294                "protocol id {:#04x} is not MQTT-SN 1.2",
295                b[1]
296            );
297            let client_id = text(&b[4..])?;
298            ensure!(
299                !client_id.is_empty() && client_id.len() <= 23,
300                "the client id is 1 to 23 characters"
301            );
302            Packet::Connect {
303                flags: Flags::decode(b[0]),
304                duration: u16_at(b, 2)?,
305                client_id,
306            }
307        }
308        0x05 => {
309            need(1)?;
310            Packet::ConnAck { rc: b[0] }
311        }
312        0x06 => Packet::WillTopicReq,
313        0x07 => {
314            if b.is_empty() {
315                Packet::WillTopic {
316                    flags: Flags::default(),
317                    topic: String::new(),
318                }
319            } else {
320                Packet::WillTopic {
321                    flags: Flags::decode(b[0]),
322                    topic: text(&b[1..])?,
323                }
324            }
325        }
326        0x08 => Packet::WillMsgReq,
327        0x09 => Packet::WillMsg { msg: b.to_vec() },
328        0x0A => {
329            need(4)?;
330            Packet::Register {
331                topic_id: u16_at(b, 0)?,
332                msg_id: u16_at(b, 2)?,
333                topic: text(&b[4..])?,
334            }
335        }
336        0x0B => {
337            need(5)?;
338            Packet::RegAck {
339                topic_id: u16_at(b, 0)?,
340                msg_id: u16_at(b, 2)?,
341                rc: b[4],
342            }
343        }
344        0x0C => {
345            need(5)?;
346            Packet::Publish {
347                flags: Flags::decode(b[0]),
348                topic: u16_at(b, 1)?,
349                msg_id: u16_at(b, 3)?,
350                data: b[5..].to_vec(),
351            }
352        }
353        0x0D => {
354            need(5)?;
355            Packet::PubAck {
356                topic_id: u16_at(b, 0)?,
357                msg_id: u16_at(b, 2)?,
358                rc: b[4],
359            }
360        }
361        0x0E => Packet::PubComp {
362            msg_id: u16_at(b, 0)?,
363        },
364        0x0F => Packet::PubRec {
365            msg_id: u16_at(b, 0)?,
366        },
367        0x10 => Packet::PubRel {
368            msg_id: u16_at(b, 0)?,
369        },
370        0x12 | 0x14 => {
371            need(3)?;
372            let flags = Flags::decode(b[0]);
373            let msg_id = u16_at(b, 1)?;
374            let (topic_name, topic_id) = if flags.topic_id_type == TOPIC_NORMAL {
375                let t = text(&b[3..])?;
376                ensure!(!t.is_empty(), "the topic name is empty");
377                (Some(t), 0)
378            } else {
379                (None, u16_at(b, 3)?)
380            };
381            if kind == 0x12 {
382                Packet::Subscribe {
383                    flags,
384                    msg_id,
385                    topic_name,
386                    topic_id,
387                }
388            } else {
389                Packet::Unsubscribe {
390                    flags,
391                    msg_id,
392                    topic_name,
393                    topic_id,
394                }
395            }
396        }
397        0x13 => {
398            need(6)?;
399            Packet::SubAck {
400                flags: Flags::decode(b[0]),
401                topic_id: u16_at(b, 1)?,
402                msg_id: u16_at(b, 3)?,
403                rc: b[5],
404            }
405        }
406        0x15 => Packet::UnsubAck {
407            msg_id: u16_at(b, 0)?,
408        },
409        0x16 => Packet::PingReq {
410            client_id: if b.is_empty() { None } else { Some(text(b)?) },
411        },
412        0x17 => Packet::PingResp,
413        0x18 => Packet::Disconnect {
414            duration: if b.is_empty() {
415                None
416            } else {
417                Some(u16_at(b, 0)?)
418            },
419        },
420        0x1A => {
421            if b.is_empty() {
422                Packet::WillTopicUpd {
423                    flags: Flags::default(),
424                    topic: String::new(),
425                }
426            } else {
427                Packet::WillTopicUpd {
428                    flags: Flags::decode(b[0]),
429                    topic: text(&b[1..])?,
430                }
431            }
432        }
433        0x1B => {
434            need(1)?;
435            Packet::WillTopicResp { rc: b[0] }
436        }
437        0x1C => Packet::WillMsgUpd { msg: b.to_vec() },
438        0x1D => {
439            need(1)?;
440            Packet::WillMsgResp { rc: b[0] }
441        }
442        other => bail!("message type {other:#04x} is not MQTT-SN"),
443    })
444}
445
446pub fn encode(p: &Packet) -> Vec<u8> {
447    let mut b: Vec<u8> = Vec::new();
448    let kind: u8 = match p {
449        Packet::Advertise { gw_id, duration } => {
450            b.push(*gw_id);
451            b.extend(duration.to_be_bytes());
452            0x00
453        }
454        Packet::SearchGw { radius } => {
455            b.push(*radius);
456            0x01
457        }
458        Packet::GwInfo { gw_id, gw_add } => {
459            b.push(*gw_id);
460            b.extend(gw_add);
461            0x02
462        }
463        Packet::Connect {
464            flags,
465            duration,
466            client_id,
467        } => {
468            b.push(flags.encode());
469            b.push(PROTOCOL_ID);
470            b.extend(duration.to_be_bytes());
471            b.extend(client_id.as_bytes());
472            0x04
473        }
474        Packet::ConnAck { rc } => {
475            b.push(*rc);
476            0x05
477        }
478        Packet::WillTopicReq => 0x06,
479        Packet::WillTopic { flags, topic } => {
480            if !topic.is_empty() {
481                b.push(flags.encode());
482                b.extend(topic.as_bytes());
483            }
484            0x07
485        }
486        Packet::WillMsgReq => 0x08,
487        Packet::WillMsg { msg } => {
488            b.extend(msg);
489            0x09
490        }
491        Packet::Register {
492            topic_id,
493            msg_id,
494            topic,
495        } => {
496            b.extend(topic_id.to_be_bytes());
497            b.extend(msg_id.to_be_bytes());
498            b.extend(topic.as_bytes());
499            0x0A
500        }
501        Packet::RegAck {
502            topic_id,
503            msg_id,
504            rc,
505        } => {
506            b.extend(topic_id.to_be_bytes());
507            b.extend(msg_id.to_be_bytes());
508            b.push(*rc);
509            0x0B
510        }
511        Packet::Publish {
512            flags,
513            topic,
514            msg_id,
515            data,
516        } => {
517            b.push(flags.encode());
518            b.extend(topic.to_be_bytes());
519            b.extend(msg_id.to_be_bytes());
520            b.extend(data);
521            0x0C
522        }
523        Packet::PubAck {
524            topic_id,
525            msg_id,
526            rc,
527        } => {
528            b.extend(topic_id.to_be_bytes());
529            b.extend(msg_id.to_be_bytes());
530            b.push(*rc);
531            0x0D
532        }
533        Packet::PubComp { msg_id } => {
534            b.extend(msg_id.to_be_bytes());
535            0x0E
536        }
537        Packet::PubRec { msg_id } => {
538            b.extend(msg_id.to_be_bytes());
539            0x0F
540        }
541        Packet::PubRel { msg_id } => {
542            b.extend(msg_id.to_be_bytes());
543            0x10
544        }
545        Packet::Subscribe {
546            flags,
547            msg_id,
548            topic_name,
549            topic_id,
550        }
551        | Packet::Unsubscribe {
552            flags,
553            msg_id,
554            topic_name,
555            topic_id,
556        } => {
557            b.push(flags.encode());
558            b.extend(msg_id.to_be_bytes());
559            match topic_name {
560                Some(t) if flags.topic_id_type == TOPIC_NORMAL => b.extend(t.as_bytes()),
561                _ => b.extend(topic_id.to_be_bytes()),
562            }
563            if matches!(p, Packet::Subscribe { .. }) {
564                0x12
565            } else {
566                0x14
567            }
568        }
569        Packet::SubAck {
570            flags,
571            topic_id,
572            msg_id,
573            rc,
574        } => {
575            b.push(flags.encode());
576            b.extend(topic_id.to_be_bytes());
577            b.extend(msg_id.to_be_bytes());
578            b.push(*rc);
579            0x13
580        }
581        Packet::UnsubAck { msg_id } => {
582            b.extend(msg_id.to_be_bytes());
583            0x15
584        }
585        Packet::PingReq { client_id } => {
586            if let Some(c) = client_id {
587                b.extend(c.as_bytes());
588            }
589            0x16
590        }
591        Packet::PingResp => 0x17,
592        Packet::Disconnect { duration } => {
593            if let Some(d) = duration {
594                b.extend(d.to_be_bytes());
595            }
596            0x18
597        }
598        Packet::WillTopicUpd { flags, topic } => {
599            if !topic.is_empty() {
600                b.push(flags.encode());
601                b.extend(topic.as_bytes());
602            }
603            0x1A
604        }
605        Packet::WillTopicResp { rc } => {
606            b.push(*rc);
607            0x1B
608        }
609        Packet::WillMsgUpd { msg } => {
610            b.extend(msg);
611            0x1C
612        }
613        Packet::WillMsgResp { rc } => {
614            b.push(*rc);
615            0x1D
616        }
617    };
618    let mut out = if b.len() + 2 <= 255 {
619        vec![(b.len() + 2) as u8]
620    } else {
621        let mut h = vec![0x01];
622        h.extend(((b.len() + 4) as u16).to_be_bytes());
623        h
624    };
625    out.push(kind);
626    out.extend(b);
627    out
628}
629
630/// A short topic name's two bytes as a u16.
631pub fn short_topic(name: &str) -> Option<u16> {
632    let b = name.as_bytes();
633    (b.len() == 2 && !name.contains(['+', '#'])).then(|| u16::from_be_bytes([b[0], b[1]]))
634}
635
636pub fn short_name(id: u16) -> String {
637    String::from_utf8_lossy(&id.to_be_bytes()).into_owned()
638}
639
640/// MQTT topic filter matching ('+' one level, '#' the rest); wildcards at the first level do
641/// not match topics starting with '$'.
642pub fn matches(filter: &str, topic: &str) -> bool {
643    if topic.starts_with('$') && filter.starts_with(['#', '+']) {
644        return false;
645    }
646    let mut f = filter.split('/');
647    let mut t = topic.split('/');
648    loop {
649        match (f.next(), t.next()) {
650            (Some("#"), _) => return true,
651            (Some("+"), Some(_)) => {}
652            (Some(a), Some(b)) if a == b => {}
653            (None, None) => return true,
654            _ => return false,
655        }
656    }
657}
658
659pub fn valid_filter(filter: &str) -> bool {
660    !filter.is_empty()
661        && filter.len() <= 1024
662        && filter.split('/').enumerate().all(|(i, level)| {
663            (level == "#" && i == filter.split('/').count() - 1)
664                || level == "+"
665                || !level.contains(['+', '#'])
666        })
667}
668
669pub fn valid_topic(topic: &str) -> bool {
670    !topic.is_empty() && topic.len() <= 1024 && !topic.contains(['+', '#', '\0'])
671}
netget/server/mqtt_sn/packet.rs

netget/server/mqtt_sn/
packet.rs
