-
Notifications
You must be signed in to change notification settings - Fork 227
/
mod.rs
501 lines (448 loc) · 15.5 KB
/
mod.rs
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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
use std::slice::Iter;
use bytes::{Buf, BufMut, Bytes, BytesMut};
mod connack;
mod connect;
mod disconnect;
mod ping;
mod puback;
mod pubcomp;
mod publish;
mod pubrec;
mod pubrel;
mod suback;
mod subscribe;
mod unsuback;
mod unsubscribe;
pub use connack::*;
pub use connect::*;
pub use disconnect::*;
pub use ping::*;
pub use puback::*;
pub use pubcomp::*;
pub use publish::*;
pub use pubrec::*;
pub use pubrel::*;
pub use suback::*;
pub use subscribe::*;
pub use unsuback::*;
pub use unsubscribe::*;
/// Encapsulates all MQTT packet types
#[derive(Debug, Clone, PartialEq)]
pub enum Packet {
Connect(Connect),
ConnAck(ConnAck),
Publish(Publish),
PubAck(PubAck),
PubRec(PubRec),
PubRel(PubRel),
PubComp(PubComp),
Subscribe(Subscribe),
SubAck(SubAck),
Unsubscribe(Unsubscribe),
UnsubAck(UnsubAck),
PingReq,
PingResp,
Disconnect(Disconnect),
}
/// MQTT packet type
#[repr(u8)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PacketType {
Connect = 1,
ConnAck,
Publish,
PubAck,
PubRec,
PubRel,
PubComp,
Subscribe,
SubAck,
Unsubscribe,
UnsubAck,
PingReq,
PingResp,
Disconnect,
}
#[repr(u8)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PropertyType {
PayloadFormatIndicator = 1,
MessageExpiryInterval = 2,
ContentType = 3,
ResponseTopic = 8,
CorrelationData = 9,
SubscriptionIdentifier = 11,
SessionExpiryInterval = 17,
AssignedClientIdentifier = 18,
ServerKeepAlive = 19,
AuthenticationMethod = 21,
AuthenticationData = 22,
RequestProblemInformation = 23,
WillDelayInterval = 24,
RequestResponseInformation = 25,
ResponseInformation = 26,
ServerReference = 28,
ReasonString = 31,
ReceiveMaximum = 33,
TopicAliasMaximum = 34,
TopicAlias = 35,
MaximumQos = 36,
RetainAvailable = 37,
UserProperty = 38,
MaximumPacketSize = 39,
WildcardSubscriptionAvailable = 40,
SubscriptionIdentifierAvailable = 41,
SharedSubscriptionAvailable = 42,
}
/// Error during serialization and deserialization
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum Error {
#[error("Expected Connect, received: {0:?}")]
NotConnect(PacketType),
#[error("Unexpected Connect")]
UnexpectedConnect,
#[error("Invalid Connect return code: {0}")]
InvalidConnectReturnCode(u8),
#[error("Invalid protocol")]
InvalidProtocol,
#[error("Invalid protocol level: {0}")]
InvalidProtocolLevel(u8),
#[error("Incorrect packet format")]
IncorrectPacketFormat,
#[error("Invalid packet type: {0}")]
InvalidPacketType(u8),
#[error("Invalid property type: {0}")]
InvalidPropertyType(u8),
#[error("Invalid QoS level: {0}")]
InvalidQoS(u8),
#[error("Invalid retain forward rule: {0}")]
InvalidRetainForwardRule(u8),
#[error("Invalid subscribe reason code: {0}")]
InvalidSubscribeReasonCode(u8),
#[error("Packet id Zero")]
PacketIdZero,
#[error("Payload size is incorrect")]
PayloadSizeIncorrect,
#[error("payload is too long")]
PayloadTooLong,
#[error("payload size limit exceeded: {0}")]
PayloadSizeLimitExceeded(usize),
#[error("Payload required")]
PayloadRequired,
#[error("Topic is not UTF-8")]
TopicNotUtf8,
#[error("Promised boundary crossed: {0}")]
BoundaryCrossed(usize),
#[error("Malformed packet")]
MalformedPacket,
#[error("Malformed remaining length")]
MalformedRemainingLength,
#[error("A Subscribe packet must contain atleast one filter")]
EmptySubscription,
/// More bytes required to frame packet. Argument
/// implies minimum additional bytes required to
/// proceed further
#[error("At least {0} more bytes required to frame packet")]
InsufficientBytes(usize),
}
/// Protocol type
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Protocol {
V4,
V5,
}
/// Quality of service
#[repr(u8)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd)]
pub enum QoS {
AtMostOnce = 0,
AtLeastOnce = 1,
ExactlyOnce = 2,
}
/// Packet type from a byte
///
/// ```ignore
/// 7 3 0
/// +--------------------------+--------------------------+
/// byte 1 | MQTT Control Packet Type | Flags for each type |
/// +--------------------------+--------------------------+
/// | Remaining Bytes Len (1/2/3/4 bytes) |
/// +-----------------------------------------------------+
///
/// http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Figure_2.2_-
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd)]
pub struct FixedHeader {
/// First byte of the stream. Used to identify packet types and
/// several flags
byte1: u8,
/// Length of fixed header. Byte 1 + (1..4) bytes. So fixed header
/// len can vary from 2 bytes to 5 bytes
/// 1..4 bytes are variable length encoded to represent remaining length
fixed_header_len: usize,
/// Remaining length of the packet. Doesn't include fixed header bytes
/// Represents variable header + payload size
remaining_len: usize,
}
impl FixedHeader {
pub fn new(byte1: u8, remaining_len_len: usize, remaining_len: usize) -> FixedHeader {
FixedHeader {
byte1,
fixed_header_len: remaining_len_len + 1,
remaining_len,
}
}
pub fn packet_type(&self) -> Result<PacketType, Error> {
let num = self.byte1 >> 4;
match num {
1 => Ok(PacketType::Connect),
2 => Ok(PacketType::ConnAck),
3 => Ok(PacketType::Publish),
4 => Ok(PacketType::PubAck),
5 => Ok(PacketType::PubRec),
6 => Ok(PacketType::PubRel),
7 => Ok(PacketType::PubComp),
8 => Ok(PacketType::Subscribe),
9 => Ok(PacketType::SubAck),
10 => Ok(PacketType::Unsubscribe),
11 => Ok(PacketType::UnsubAck),
12 => Ok(PacketType::PingReq),
13 => Ok(PacketType::PingResp),
14 => Ok(PacketType::Disconnect),
_ => Err(Error::InvalidPacketType(num)),
}
}
/// Returns the size of full packet (fixed header + variable header + payload)
/// Fixed header is enough to get the size of a frame in the stream
pub fn frame_length(&self) -> usize {
self.fixed_header_len + self.remaining_len
}
}
fn property(num: u8) -> Result<PropertyType, Error> {
let property = match num {
1 => PropertyType::PayloadFormatIndicator,
2 => PropertyType::MessageExpiryInterval,
3 => PropertyType::ContentType,
8 => PropertyType::ResponseTopic,
9 => PropertyType::CorrelationData,
11 => PropertyType::SubscriptionIdentifier,
17 => PropertyType::SessionExpiryInterval,
18 => PropertyType::AssignedClientIdentifier,
19 => PropertyType::ServerKeepAlive,
21 => PropertyType::AuthenticationMethod,
22 => PropertyType::AuthenticationData,
23 => PropertyType::RequestProblemInformation,
24 => PropertyType::WillDelayInterval,
25 => PropertyType::RequestResponseInformation,
26 => PropertyType::ResponseInformation,
28 => PropertyType::ServerReference,
31 => PropertyType::ReasonString,
33 => PropertyType::ReceiveMaximum,
34 => PropertyType::TopicAliasMaximum,
35 => PropertyType::TopicAlias,
36 => PropertyType::MaximumQos,
37 => PropertyType::RetainAvailable,
38 => PropertyType::UserProperty,
39 => PropertyType::MaximumPacketSize,
40 => PropertyType::WildcardSubscriptionAvailable,
41 => PropertyType::SubscriptionIdentifierAvailable,
42 => PropertyType::SharedSubscriptionAvailable,
num => return Err(Error::InvalidPropertyType(num)),
};
Ok(property)
}
/// Checks if the stream has enough bytes to frame a packet and returns fixed header
/// only if a packet can be framed with existing bytes in the `stream`.
/// The passed stream doesn't modify parent stream's cursor. If this function
/// returned an error, next `check` on the same parent stream is forced start
/// with cursor at 0 again (Iter is owned. Only Iter's cursor is changed internally)
pub fn check(stream: Iter<u8>, max_packet_size: usize) -> Result<FixedHeader, Error> {
// Create fixed header if there are enough bytes in the stream
// to frame full packet
let stream_len = stream.len();
let fixed_header = parse_fixed_header(stream)?;
// Don't let rogue connections attack with huge payloads.
// Disconnect them before reading all that data
if fixed_header.remaining_len > max_packet_size {
return Err(Error::PayloadSizeLimitExceeded(fixed_header.remaining_len));
}
// If the current call fails due to insufficient bytes in the stream,
// after calculating remaining length, we extend the stream
let frame_length = fixed_header.frame_length();
if stream_len < frame_length {
return Err(Error::InsufficientBytes(frame_length - stream_len));
}
Ok(fixed_header)
}
/// Parses fixed header
fn parse_fixed_header(mut stream: Iter<u8>) -> Result<FixedHeader, Error> {
// At least 2 bytes are necessary to frame a packet
let stream_len = stream.len();
if stream_len < 2 {
return Err(Error::InsufficientBytes(2 - stream_len));
}
let byte1 = stream.next().unwrap();
let (len_len, len) = length(stream)?;
Ok(FixedHeader::new(*byte1, len_len, len))
}
/// Parses variable byte integer in the stream and returns the length
/// and number of bytes that make it. Used for remaining length calculation
/// as well as for calculating property lengths
fn length(stream: Iter<u8>) -> Result<(usize, usize), Error> {
let mut len: usize = 0;
let mut len_len = 0;
let mut done = false;
let mut shift = 0;
// Use continuation bit at position 7 to continue reading next
// byte to frame 'length'.
// Stream 0b1xxx_xxxx 0b1yyy_yyyy 0b1zzz_zzzz 0b0www_wwww will
// be framed as number 0bwww_wwww_zzz_zzzz_yyy_yyyy_xxx_xxxx
for byte in stream {
len_len += 1;
let byte = *byte as usize;
len += (byte & 0x7F) << shift;
// stop when continue bit is 0
done = (byte & 0x80) == 0;
if done {
break;
}
shift += 7;
// Only a max of 4 bytes allowed for remaining length
// more than 4 shifts (0, 7, 14, 21) implies bad length
if shift > 21 {
return Err(Error::MalformedRemainingLength);
}
}
// Not enough bytes to frame remaining length. wait for
// one more byte
if !done {
return Err(Error::InsufficientBytes(1));
}
Ok((len_len, len))
}
/// Reads a stream of bytes and extracts next MQTT packet out of it
pub fn read(stream: &mut BytesMut, max_size: usize) -> Result<Packet, Error> {
let fixed_header = check(stream.iter(), max_size)?;
// Test with a stream with exactly the size to check border panics
let packet = stream.split_to(fixed_header.frame_length());
let packet_type = fixed_header.packet_type()?;
if fixed_header.remaining_len == 0 {
// no payload packets
return match packet_type {
PacketType::PingReq => Ok(Packet::PingReq),
PacketType::PingResp => Ok(Packet::PingResp),
_ => Err(Error::PayloadRequired),
};
}
let packet = packet.freeze();
let packet = match packet_type {
PacketType::Connect => Packet::Connect(Connect::read(fixed_header, packet)?),
PacketType::ConnAck => Packet::ConnAck(ConnAck::read(fixed_header, packet)?),
PacketType::Publish => Packet::Publish(Publish::read(fixed_header, packet)?),
PacketType::PubAck => Packet::PubAck(PubAck::read(fixed_header, packet)?),
PacketType::PubRec => Packet::PubRec(PubRec::read(fixed_header, packet)?),
PacketType::PubRel => Packet::PubRel(PubRel::read(fixed_header, packet)?),
PacketType::PubComp => Packet::PubComp(PubComp::read(fixed_header, packet)?),
PacketType::Subscribe => Packet::Subscribe(Subscribe::read(fixed_header, packet)?),
PacketType::SubAck => Packet::SubAck(SubAck::read(fixed_header, packet)?),
PacketType::Unsubscribe => Packet::Unsubscribe(Unsubscribe::read(fixed_header, packet)?),
PacketType::UnsubAck => Packet::UnsubAck(UnsubAck::read(fixed_header, packet)?),
PacketType::PingReq => Packet::PingReq,
PacketType::PingResp => Packet::PingResp,
PacketType::Disconnect => Packet::Disconnect(Disconnect::read(fixed_header, packet)?),
};
Ok(packet)
}
/// Reads a series of bytes with a length from a byte stream
fn read_mqtt_bytes(stream: &mut Bytes) -> Result<Bytes, Error> {
let len = read_u16(stream)? as usize;
// Prevent attacks with wrong remaining length. This method is used in
// `packet.assembly()` with (enough) bytes to frame packet. Ensures that
// reading variable len string or bytes doesn't cross promised boundary
// with `read_fixed_header()`
if len > stream.len() {
return Err(Error::BoundaryCrossed(len));
}
Ok(stream.split_to(len))
}
/// Reads a string from bytes stream
fn read_mqtt_string(stream: &mut Bytes) -> Result<String, Error> {
let s = read_mqtt_bytes(stream)?;
match String::from_utf8(s.to_vec()) {
Ok(v) => Ok(v),
Err(_e) => Err(Error::TopicNotUtf8),
}
}
/// Serializes bytes to stream (including length)
fn write_mqtt_bytes(stream: &mut BytesMut, bytes: &[u8]) {
stream.put_u16(bytes.len() as u16);
stream.extend_from_slice(bytes);
}
/// Serializes a string to stream
fn write_mqtt_string(stream: &mut BytesMut, string: &str) {
write_mqtt_bytes(stream, string.as_bytes());
}
/// Writes remaining length to stream and returns number of bytes for remaining length
fn write_remaining_length(stream: &mut BytesMut, len: usize) -> Result<usize, Error> {
if len > 268_435_455 {
return Err(Error::PayloadTooLong);
}
let mut done = false;
let mut x = len;
let mut count = 0;
while !done {
let mut byte = (x % 128) as u8;
x /= 128;
if x > 0 {
byte |= 128;
}
stream.put_u8(byte);
count += 1;
done = x == 0;
}
Ok(count)
}
/// Return number of remaining length bytes required for encoding length
fn len_len(len: usize) -> usize {
if len >= 2_097_152 {
4
} else if len >= 16_384 {
3
} else if len >= 128 {
2
} else {
1
}
}
/// Maps a number to QoS
pub fn qos(num: u8) -> Result<QoS, Error> {
match num {
0 => Ok(QoS::AtMostOnce),
1 => Ok(QoS::AtLeastOnce),
2 => Ok(QoS::ExactlyOnce),
qos => Err(Error::InvalidQoS(qos)),
}
}
/// After collecting enough bytes to frame a packet (packet's frame())
/// , It's possible that content itself in the stream is wrong. Like expected
/// packet id or qos not being present. In cases where `read_mqtt_string` or
/// `read_mqtt_bytes` exhausted remaining length but packet framing expects to
/// parse qos next, these pre checks will prevent `bytes` crashes
fn read_u16(stream: &mut Bytes) -> Result<u16, Error> {
if stream.len() < 2 {
return Err(Error::MalformedPacket);
}
Ok(stream.get_u16())
}
fn read_u8(stream: &mut Bytes) -> Result<u8, Error> {
if stream.is_empty() {
return Err(Error::MalformedPacket);
}
Ok(stream.get_u8())
}
fn read_u32(stream: &mut Bytes) -> Result<u32, Error> {
if stream.len() < 4 {
return Err(Error::MalformedPacket);
}
Ok(stream.get_u32())
}