use serde_json::{json, Map, Value};
pub const PAYLOAD_FORMAT_INDICATOR: u8 = 0x01;
pub const MESSAGE_EXPIRY_INTERVAL: u8 = 0x02;
pub const CONTENT_TYPE: u8 = 0x03;
pub const RESPONSE_TOPIC: u8 = 0x08;
pub const CORRELATION_DATA: u8 = 0x09;
pub const SUBSCRIPTION_IDENTIFIER: u8 = 0x0B;
pub const SESSION_EXPIRY_INTERVAL: u8 = 0x11;
pub const ASSIGNED_CLIENT_IDENTIFIER: u8 = 0x12;
pub const SERVER_KEEP_ALIVE: u8 = 0x13;
pub const AUTHENTICATION_METHOD: u8 = 0x15;
pub const AUTHENTICATION_DATA: u8 = 0x16;
pub const REQUEST_PROBLEM_INFORMATION: u8 = 0x17;
pub const WILL_DELAY_INTERVAL: u8 = 0x18;
pub const REQUEST_RESPONSE_INFORMATION: u8 = 0x19;
pub const RESPONSE_INFORMATION: u8 = 0x1A;
pub const SERVER_REFERENCE: u8 = 0x1C;
pub const REASON_STRING: u8 = 0x1F;
pub const RECEIVE_MAXIMUM: u8 = 0x21;
pub const TOPIC_ALIAS_MAXIMUM: u8 = 0x22;
pub const TOPIC_ALIAS: u8 = 0x23;
pub const MAXIMUM_QOS: u8 = 0x24;
pub const RETAIN_AVAILABLE: u8 = 0x25;
pub const USER_PROPERTY: u8 = 0x26;
pub const MAXIMUM_PACKET_SIZE: u8 = 0x27;
pub const WILDCARD_SUBSCRIPTION_AVAILABLE: u8 = 0x28;
pub const SUBSCRIPTION_IDENTIFIER_AVAILABLE: u8 = 0x29;
pub const SHARED_SUBSCRIPTION_AVAILABLE: u8 = 0x2A;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Kind {
Byte,
U16,
U32,
Varint,
Utf8,
Binary,
Pair,
}
fn describe(id: u8) -> Option<(&'static str, Kind)> {
Some(match id {
PAYLOAD_FORMAT_INDICATOR => ("payload_format_indicator", Kind::Byte),
MESSAGE_EXPIRY_INTERVAL => ("message_expiry_interval", Kind::U32),
CONTENT_TYPE => ("content_type", Kind::Utf8),
RESPONSE_TOPIC => ("response_topic", Kind::Utf8),
CORRELATION_DATA => ("correlation_data", Kind::Binary),
SUBSCRIPTION_IDENTIFIER => ("subscription_identifier", Kind::Varint),
SESSION_EXPIRY_INTERVAL => ("session_expiry_interval", Kind::U32),
ASSIGNED_CLIENT_IDENTIFIER => ("assigned_client_identifier", Kind::Utf8),
SERVER_KEEP_ALIVE => ("server_keep_alive", Kind::U16),
AUTHENTICATION_METHOD => ("authentication_method", Kind::Utf8),
AUTHENTICATION_DATA => ("authentication_data", Kind::Binary),
REQUEST_PROBLEM_INFORMATION => ("request_problem_information", Kind::Byte),
WILL_DELAY_INTERVAL => ("will_delay_interval", Kind::U32),
REQUEST_RESPONSE_INFORMATION => ("request_response_information", Kind::Byte),
RESPONSE_INFORMATION => ("response_information", Kind::Utf8),
SERVER_REFERENCE => ("server_reference", Kind::Utf8),
REASON_STRING => ("reason_string", Kind::Utf8),
RECEIVE_MAXIMUM => ("receive_maximum", Kind::U16),
TOPIC_ALIAS_MAXIMUM => ("topic_alias_maximum", Kind::U16),
TOPIC_ALIAS => ("topic_alias", Kind::U16),
MAXIMUM_QOS => ("maximum_qos", Kind::Byte),
RETAIN_AVAILABLE => ("retain_available", Kind::Byte),
USER_PROPERTY => ("user_properties", Kind::Pair),
MAXIMUM_PACKET_SIZE => ("maximum_packet_size", Kind::U32),
WILDCARD_SUBSCRIPTION_AVAILABLE => ("wildcard_subscription_available", Kind::Byte),
SUBSCRIPTION_IDENTIFIER_AVAILABLE => ("subscription_identifier_available", Kind::Byte),
SHARED_SUBSCRIPTION_AVAILABLE => ("shared_subscription_available", Kind::Byte),
_ => return None,
})
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum PropValue {
Int(u32),
Text(String),
Binary(Vec<u8>),
Pair(String, String),
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Properties {
pub entries: Vec<(u8, PropValue)>,
}
impl Properties {
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
pub fn int(&self, id: u8) -> Option<u32> {
self.entries.iter().find_map(|(i, v)| match v {
PropValue::Int(n) if *i == id => Some(*n),
_ => None,
})
}
pub fn text(&self, id: u8) -> Option<&str> {
self.entries.iter().find_map(|(i, v)| match v {
PropValue::Text(s) if *i == id => Some(s.as_str()),
_ => None,
})
}
pub fn binary(&self, id: u8) -> Option<&[u8]> {
self.entries.iter().find_map(|(i, v)| match v {
PropValue::Binary(b) if *i == id => Some(b.as_slice()),
_ => None,
})
}
pub fn to_json(&self) -> Value {
let mut out = Map::new();
let mut pairs: Vec<Value> = Vec::new();
let mut sub_ids: Vec<Value> = Vec::new();
for (id, value) in &self.entries {
let Some((name, _)) = describe(*id) else {
continue;
};
match value {
PropValue::Pair(k, v) => pairs.push(json!([k, v])),
PropValue::Int(n) if *id == SUBSCRIPTION_IDENTIFIER => sub_ids.push(json!(n)),
PropValue::Int(n) => {
out.insert(name.to_string(), json!(n));
}
PropValue::Text(s) => {
out.insert(name.to_string(), json!(s));
}
PropValue::Binary(b) => {
let (text, encoding) = binary_to_text(b);
out.insert(name.to_string(), json!(text));
out.insert(format!("{name}_encoding"), json!(encoding));
}
}
}
if !pairs.is_empty() {
out.insert("user_properties".into(), Value::Array(pairs));
}
if !sub_ids.is_empty() {
out.insert("subscription_identifiers".into(), Value::Array(sub_ids));
}
Value::Object(out)
}
}
pub fn binary_to_text(bytes: &[u8]) -> (String, &'static str) {
match std::str::from_utf8(bytes) {
Ok(s) if !crate::utils::sanitize::has_controls(s) => (s.to_string(), "utf8"),
_ => (hex::encode(bytes), "hex"),
}
}
pub fn text_to_binary(text: &str, encoding: Option<&str>) -> Result<Vec<u8>, String> {
match encoding.unwrap_or("utf8") {
"utf8" => Ok(text.as_bytes().to_vec()),
"hex" => hex::decode(text).map_err(|e| format!("not valid hex: {e}")),
other => Err(format!(
"encoding must be \"utf8\" or \"hex\", not {other:?}"
)),
}
}
pub fn read_varint(buf: &[u8], pos: &mut usize) -> Option<u32> {
let mut value: u32 = 0;
for i in 0..4 {
let byte = *buf.get(*pos + i)?;
value |= u32::from(byte & 0x7F) << (7 * i);
if byte & 0x80 == 0 {
*pos += i + 1;
return Some(value);
}
}
None
}
pub fn write_varint(out: &mut Vec<u8>, mut value: u32) {
for _ in 0..4 {
let mut byte = (value % 128) as u8;
value /= 128;
if value > 0 {
byte |= 0x80;
}
out.push(byte);
if value == 0 {
break;
}
}
}
pub const VARINT_MAX: u32 = 268_435_455;
fn read_u16(buf: &[u8], pos: &mut usize) -> Option<u16> {
let b = buf.get(*pos..*pos + 2)?;
*pos += 2;
Some(u16::from_be_bytes([b[0], b[1]]))
}
fn read_bin(buf: &[u8], pos: &mut usize) -> Option<Vec<u8>> {
let len = usize::from(read_u16(buf, pos)?);
let end = pos.checked_add(len)?;
let b = buf.get(*pos..end)?.to_vec();
*pos = end;
Some(b)
}
fn read_utf8(buf: &[u8], pos: &mut usize) -> Option<String> {
String::from_utf8(read_bin(buf, pos)?).ok()
}
pub const MAX_USER_PROPERTIES: usize = 128;
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum PropertyError {
Malformed(String),
Duplicate(&'static str),
Protocol(String),
}
impl std::fmt::Display for PropertyError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
PropertyError::Malformed(m) => write!(f, "malformed properties: {m}"),
PropertyError::Duplicate(n) => write!(f, "property {n} appears more than once"),
PropertyError::Protocol(m) => write!(f, "invalid properties: {m}"),
}
}
}
pub fn forbid(props: &Properties, ids: &[u8], packet: &str) -> Result<(), PropertyError> {
match props.entries.iter().find(|(id, _)| ids.contains(id)) {
Some((id, _)) => Err(PropertyError::Protocol(format!(
"{} is not allowed in {packet}",
describe(*id).map(|d| d.0).unwrap_or("property")
))),
None => Ok(()),
}
}
pub fn read_properties(buf: &[u8], pos: &mut usize) -> Result<Properties, PropertyError> {
let malformed = |m: &str| PropertyError::Malformed(m.to_string());
let len = read_varint(buf, pos).ok_or_else(|| malformed("property length"))? as usize;
let end = pos
.checked_add(len)
.filter(|e| *e <= buf.len())
.ok_or_else(|| malformed("property length runs past the packet"))?;
let block = &buf[*pos..end];
let mut p = 0usize;
let mut props = Properties::default();
let mut user_properties = 0usize;
while p < block.len() {
let id = read_varint(block, &mut p).ok_or_else(|| malformed("property identifier"))?;
let id = u8::try_from(id).map_err(|_| malformed("property identifier out of range"))?;
let (name, kind) =
describe(id).ok_or_else(|| malformed(&format!("unknown property 0x{id:02x}")))?;
let value = match kind {
Kind::Byte => PropValue::Int(u32::from(*block.get(p).ok_or_else(|| malformed(name))?)),
Kind::U16 => PropValue::Int(u32::from(
read_u16(block, &mut p).ok_or_else(|| malformed(name))?,
)),
Kind::U32 => {
let b = block.get(p..p + 4).ok_or_else(|| malformed(name))?;
p += 4;
PropValue::Int(u32::from_be_bytes([b[0], b[1], b[2], b[3]]))
}
Kind::Varint => {
PropValue::Int(read_varint(block, &mut p).ok_or_else(|| malformed(name))?)
}
Kind::Utf8 => PropValue::Text(read_utf8(block, &mut p).ok_or_else(|| malformed(name))?),
Kind::Binary => {
PropValue::Binary(read_bin(block, &mut p).ok_or_else(|| malformed(name))?)
}
Kind::Pair => {
let k = read_utf8(block, &mut p).ok_or_else(|| malformed(name))?;
let v = read_utf8(block, &mut p).ok_or_else(|| malformed(name))?;
PropValue::Pair(k, v)
}
};
if kind == Kind::Byte {
p += 1;
}
if id == USER_PROPERTY {
user_properties += 1;
if user_properties > MAX_USER_PROPERTIES {
return Err(PropertyError::Protocol(format!(
"more than {MAX_USER_PROPERTIES} user properties"
)));
}
} else if props.entries.iter().any(|(i, _)| *i == id) {
return Err(PropertyError::Duplicate(name));
}
if id == SUBSCRIPTION_IDENTIFIER && value == PropValue::Int(0) {
return Err(PropertyError::Protocol(
"subscription identifier 0".to_string(),
));
}
props.entries.push((id, value));
}
*pos = end;
Ok(props)
}
#[derive(Default, Debug, Clone)]
pub struct PropertyWriter {
body: Vec<u8>,
}
impl PropertyWriter {
pub fn new() -> Self {
Self::default()
}
pub fn is_empty(&self) -> bool {
self.body.is_empty()
}
pub fn byte(&mut self, id: u8, v: u8) -> &mut Self {
self.body.extend_from_slice(&[id, v]);
self
}
pub fn u16(&mut self, id: u8, v: u16) -> &mut Self {
self.body.push(id);
self.body.extend_from_slice(&v.to_be_bytes());
self
}
pub fn u32(&mut self, id: u8, v: u32) -> &mut Self {
self.body.push(id);
self.body.extend_from_slice(&v.to_be_bytes());
self
}
pub fn varint(&mut self, id: u8, v: u32) -> &mut Self {
self.body.push(id);
write_varint(&mut self.body, v);
self
}
fn lp(&mut self, bytes: &[u8]) {
let len = bytes.len().min(usize::from(u16::MAX));
self.body.extend_from_slice(&(len as u16).to_be_bytes());
self.body.extend_from_slice(&bytes[..len]);
}
pub fn text(&mut self, id: u8, v: &str) -> &mut Self {
self.body.push(id);
self.lp(v.as_bytes());
self
}
pub fn binary(&mut self, id: u8, v: &[u8]) -> &mut Self {
self.body.push(id);
self.lp(v);
self
}
pub fn pair(&mut self, k: &str, v: &str) -> &mut Self {
self.body.push(USER_PROPERTY);
self.lp(k.as_bytes());
self.lp(v.as_bytes());
self
}
pub fn extend(&mut self, props: &Properties) -> &mut Self {
for (id, value) in &props.entries {
match (describe(*id).map(|d| d.1), value) {
(Some(Kind::Byte), PropValue::Int(n)) => self.byte(*id, *n as u8),
(Some(Kind::U16), PropValue::Int(n)) => self.u16(*id, *n as u16),
(Some(Kind::U32), PropValue::Int(n)) => self.u32(*id, *n),
(Some(Kind::Varint), PropValue::Int(n)) => self.varint(*id, *n),
(Some(Kind::Utf8), PropValue::Text(s)) => self.text(*id, s),
(Some(Kind::Binary), PropValue::Binary(b)) => self.binary(*id, b),
(Some(Kind::Pair), PropValue::Pair(k, v)) => self.pair(k, v),
_ => self,
};
}
self
}
pub fn append(&mut self, other: &PropertyWriter) -> &mut Self {
self.body.extend_from_slice(&other.body);
self
}
pub fn finish(&self) -> Vec<u8> {
let mut out = Vec::with_capacity(self.body.len() + 4);
write_varint(&mut out, self.body.len() as u32);
out.extend_from_slice(&self.body);
out
}
}
pub fn user_properties_from_json(v: Option<&Value>) -> Result<Vec<(String, String)>, String> {
let Some(v) = v else {
return Ok(Vec::new());
};
let as_text = |x: &Value| -> Result<String, String> {
match x {
Value::String(s) => Ok(s.clone()),
Value::Number(n) => Ok(n.to_string()),
Value::Bool(b) => Ok(b.to_string()),
other => Err(format!("user property values must be strings, not {other}")),
}
};
let pairs = match v {
Value::Null => Vec::new(),
Value::Object(map) => map
.iter()
.map(|(k, x)| Ok((k.clone(), as_text(x)?)))
.collect::<Result<Vec<_>, String>>()?,
Value::Array(items) => items
.iter()
.map(|item| match item {
Value::Array(kv) if kv.len() == 2 => Ok((as_text(&kv[0])?, as_text(&kv[1])?)),
Value::Object(o) => match (o.get("name").or(o.get("key")), o.get("value")) {
(Some(k), Some(x)) => Ok((as_text(k)?, as_text(x)?)),
_ => Err("each user property object needs \"name\" and \"value\"".into()),
},
other => Err(format!(
"each user property is a [name, value] pair, not {other}"
)),
})
.collect::<Result<Vec<_>, String>>()?,
other => {
return Err(format!(
"user_properties is a list of [name, value] pairs or an object, not {other}"
))
}
};
for (k, x) in &pairs {
if k.len() > usize::from(u16::MAX) || x.len() > usize::from(u16::MAX) {
return Err("a user property name or value is longer than 65535 bytes".into());
}
}
Ok(pairs)
}
pub const CONNACK_REASON_CODES: &[(u8, &str)] = &[
(0x00, "success"),
(0x80, "unspecified error"),
(0x81, "malformed packet"),
(0x82, "protocol error"),
(0x83, "implementation specific error"),
(0x84, "unsupported protocol version"),
(0x85, "client identifier not valid"),
(0x86, "bad user name or password"),
(0x87, "not authorized"),
(0x88, "server unavailable"),
(0x89, "server busy"),
(0x8A, "banned"),
(0x8C, "bad authentication method"),
(0x90, "topic name invalid"),
(0x95, "packet too large"),
(0x97, "quota exceeded"),
(0x99, "payload format invalid"),
(0x9A, "retain not supported"),
(0x9B, "QoS not supported"),
(0x9C, "use another server"),
(0x9D, "server moved"),
(0x9F, "connection rate exceeded"),
];
pub const SUBACK_REASON_CODES: &[(u8, &str)] = &[
(0x00, "granted QoS 0"),
(0x01, "granted QoS 1"),
(0x02, "granted QoS 2"),
(0x80, "unspecified error"),
(0x83, "implementation specific error"),
(0x87, "not authorized"),
(0x8F, "topic filter invalid"),
(0x91, "packet identifier in use"),
(0x97, "quota exceeded"),
(0x9E, "shared subscriptions not supported"),
(0xA1, "subscription identifiers not supported"),
(0xA2, "wildcard subscriptions not supported"),
];
pub const PUBACK_REASON_CODES: &[(u8, &str)] = &[
(0x00, "success"),
(0x10, "no matching subscribers"),
(0x80, "unspecified error"),
(0x83, "implementation specific error"),
(0x87, "not authorized"),
(0x90, "topic name invalid"),
(0x91, "packet identifier in use"),
(0x97, "quota exceeded"),
(0x99, "payload format invalid"),
];
pub const UNSUBACK_REASON_CODES: &[(u8, &str)] = &[
(0x00, "success"),
(0x11, "no subscription existed"),
(0x80, "unspecified error"),
(0x83, "implementation specific error"),
(0x87, "not authorized"),
(0x8F, "topic filter invalid"),
(0x91, "packet identifier in use"),
];
pub const DISCONNECT_REASON_CODES: &[(u8, &str)] = &[
(0x00, "normal disconnection"),
(0x80, "unspecified error"),
(0x81, "malformed packet"),
(0x82, "protocol error"),
(0x83, "implementation specific error"),
(0x87, "not authorized"),
(0x89, "server busy"),
(0x8B, "server shutting down"),
(0x8D, "keep alive timeout"),
(0x8E, "session taken over"),
(0x8F, "topic filter invalid"),
(0x90, "topic name invalid"),
(0x93, "receive maximum exceeded"),
(0x94, "topic alias invalid"),
(0x95, "packet too large"),
(0x96, "message rate too high"),
(0x97, "quota exceeded"),
(0x98, "administrative action"),
(0x99, "payload format invalid"),
(0x9A, "retain not supported"),
(0x9B, "QoS not supported"),
(0x9C, "use another server"),
(0x9D, "server moved"),
(0x9E, "shared subscriptions not supported"),
(0x9F, "connection rate exceeded"),
(0xA0, "maximum connect time"),
(0xA1, "subscription identifiers not supported"),
(0xA2, "wildcard subscriptions not supported"),
];
pub fn check_reason(code: u64, table: &[(u8, &str)], packet: &str) -> Result<u8, String> {
if let Ok(c) = u8::try_from(code) {
if table.iter().any(|(t, _)| *t == c) {
return Ok(c);
}
}
let valid: Vec<String> = table
.iter()
.map(|(c, n)| format!("{c} (0x{c:02x}, {n})"))
.collect();
Err(format!(
"{code} is not an MQTT 5 {packet} reason code; use one of: {}",
valid.join(", ")
))
}
pub fn reason_name(code: u8, table: &[(u8, &'static str)]) -> &'static str {
table
.iter()
.find(|(c, _)| *c == code)
.map(|(_, n)| *n)
.unwrap_or("unknown")
}
pub fn connack_v311_to_v5(code: u8) -> u8 {
match code {
0 => 0x00,
1 => 0x84,
2 => 0x85,
3 => 0x88,
4 => 0x86,
_ => 0x87,
}
}
pub fn connack_v5_to_v311(code: u8) -> u8 {
match code {
0x00 => 0,
0x84 => 1,
0x85 => 2,
0x86 | 0x8C => 4,
0x80 | 0x83 | 0x88 | 0x89 | 0x97 | 0x9C | 0x9D | 0x9F => 3,
_ => 5,
}
}