Coverage for zenmqtt/mqtt/properties.py: 96%

1import struct 

2from enum import IntEnum 

3from typing import Any, Callable, Final, Sequence, Tuple, TypedDict, Union, cast 

4 

5from zenmqtt.exceptions import MalformedPacketError 

6from zenmqtt.mqtt.packet import BytesReader, PacketType 

7from zenmqtt.mqtt.utils import pack_binary, pack_str16, pack_variable_byte_integer 

8 

9 

10class Property(IntEnum): 

11 PAYLOAD_FORMAT_INDICATOR = 0x01 

12 MESSAGE_EXPIRY_INTERVAL = 0x02 

13 CONTENT_TYPE = 0x03 

14 

15 RESPONSE_TOPIC = 0x08 

16 CORRELATION_DATA = 0x09 

17 

18 SUBSCRIPTION_IDENTIFIER = 0x0B 

19 

20 SESSION_EXPIRY_INTERVAL = 0x11 

21 ASSIGNED_CLIENT_IDENTIFIER = 0x12 

22 

23 SERVER_KEEP_ALIVE = 0x13 

24 

25 AUTHENTICATION_METHOD = 0x15 

26 AUTHENTICATION_DATA = 0x16 

27 REQUEST_PROBLEM_INFORMATION = 0x17 

28 

29 WILL_DELAY_INTERVAL = 0x18 

30 

31 REQUEST_RESPONSE_INFORMATION = 0x19 

32 RESPONSE_INFORMATION = 0x1A 

33 

34 SERVER_REFERENCE = 0x1C 

35 

36 REASON_STRING = 0x1F 

37 

38 RECEIVE_MAXIMUM = 0x21 

39 TOPIC_ALIAS_MAXIMUM = 0x22 

40 

41 TOPIC_ALIAS = 0x23 

42 

43 MAXIMUM_QOS = 0x24 

44 RETAIN_AVAILABLE = 0x25 

45 

46 USER_PROPERTY = 0x26 

47 

48 MAXIMUM_PACKET_SIZE = 0x27 

49 

50 WILDCARD_SUBSCRIPTION_AVAILABLE = 0x28 

51 SUBSCRIPTION_IDENTIFIER_AVAILABLE = 0x29 

52 SHARED_SUBSCRIPTION_AVAILABLE = 0x2A 

53 

54 

55# names of properties are the names of Property members in lower case 

56_NAME_TO_CODE_MAP: dict[str, Property] = {prop.name.lower(): prop for prop in Property} 

57 

58 

59_AVAILABLE_PROPERTIES_PER_TYPE = { 

60 PacketType.CONNECT: { 

61 Property.SESSION_EXPIRY_INTERVAL, 

62 Property.AUTHENTICATION_METHOD, 

63 Property.AUTHENTICATION_DATA, 

64 Property.REQUEST_PROBLEM_INFORMATION, 

65 Property.REQUEST_RESPONSE_INFORMATION, 

66 Property.RECEIVE_MAXIMUM, 

67 Property.TOPIC_ALIAS_MAXIMUM, 

68 Property.USER_PROPERTY, 

69 Property.MAXIMUM_PACKET_SIZE, 

70 }, 

71 PacketType.CONNACK: { 

72 Property.SESSION_EXPIRY_INTERVAL, 

73 Property.ASSIGNED_CLIENT_IDENTIFIER, 

74 Property.SERVER_KEEP_ALIVE, 

75 Property.AUTHENTICATION_METHOD, 

76 Property.AUTHENTICATION_DATA, 

77 Property.RESPONSE_INFORMATION, 

78 Property.SERVER_REFERENCE, 

79 Property.REASON_STRING, 

80 Property.RECEIVE_MAXIMUM, 

81 Property.TOPIC_ALIAS_MAXIMUM, 

82 Property.MAXIMUM_QOS, 

83 Property.RETAIN_AVAILABLE, 

84 Property.USER_PROPERTY, 

85 Property.MAXIMUM_PACKET_SIZE, 

86 Property.WILDCARD_SUBSCRIPTION_AVAILABLE, 

87 Property.SUBSCRIPTION_IDENTIFIER_AVAILABLE, 

88 Property.SHARED_SUBSCRIPTION_AVAILABLE, 

89 }, 

90 PacketType.PUBLISH: { 

91 Property.PAYLOAD_FORMAT_INDICATOR, 

92 Property.MESSAGE_EXPIRY_INTERVAL, 

93 Property.CONTENT_TYPE, 

94 Property.RESPONSE_TOPIC, 

95 Property.SUBSCRIPTION_IDENTIFIER, 

96 Property.TOPIC_ALIAS, 

97 Property.USER_PROPERTY, 

98 }, 

99 PacketType.PUBACK: { 

100 Property.REASON_STRING, 

101 Property.USER_PROPERTY, 

102 }, 

103 PacketType.PUBREC: { 

104 Property.REASON_STRING, 

105 Property.USER_PROPERTY, 

106 }, 

107 PacketType.PUBREL: {Property.REASON_STRING, Property.USER_PROPERTY}, 

108 PacketType.PUBCOMP: {Property.REASON_STRING, Property.USER_PROPERTY}, 

109 PacketType.SUBSCRIBE: {Property.SUBSCRIPTION_IDENTIFIER, Property.USER_PROPERTY}, 

110 PacketType.SUBACK: {Property.REASON_STRING, Property.USER_PROPERTY}, 

111 PacketType.UNSUBSCRIBE: {Property.USER_PROPERTY}, 

112 PacketType.UNSUBACK: {Property.REASON_STRING, Property.USER_PROPERTY}, 

113 PacketType.PINGREQ: {}, 

114 PacketType.PINGRESP: {}, 

115 PacketType.DISCONNECT: { 

116 Property.SESSION_EXPIRY_INTERVAL, 

117 Property.SERVER_REFERENCE, 

118 Property.REASON_STRING, 

119 Property.USER_PROPERTY, 

120 }, 

121 PacketType.AUTH: { 

122 Property.AUTHENTICATION_METHOD, 

123 Property.AUTHENTICATION_DATA, 

124 Property.REASON_STRING, 

125 Property.USER_PROPERTY, 

126 }, 

127} 

128 

129 

130class Properties(TypedDict, total=False): 

131 payload_format_indicator: bool 

132 message_expiry_interval: int 

133 content_type: str 

134 response_topic: str 

135 correlation_data: bytes 

136 # one in SUBSCRIBE, one or more in PUBLISH (MQTT 5, 3.3.2.3.8) 

137 subscription_identifier: Union[int, Sequence[int]] 

138 session_expiry_interval: int 

139 assigned_client_identifier: str 

140 server_keep_alive: int 

141 authentication_method: str 

142 authentication_data: bytes 

143 request_problem_information: bool 

144 will_delay_interval: int 

145 request_response_information: bool 

146 response_information: str 

147 server_reference: str 

148 reason_string: str 

149 receive_maximum: int 

150 topic_alias_maximum: int 

151 topic_alias: int 

152 maximum_qos: int 

153 retain_available: bool 

154 user_property: Sequence[Tuple[str, str]] 

155 maximum_packet_size: int 

156 wildcard_subscription_available: bool 

157 subscription_identifier_available: bool 

158 shared_subscription_available: bool 

159 

160 

161def _read_bool(reader: BytesReader) -> bool: 

162 return bool(reader.read_byte()) 

163 

164 

165def _read_user_property(reader: BytesReader) -> Tuple[str, str]: 

166 return reader.read_str(), reader.read_str() 

167 

168 

169_MAP_PROPERTY_PARSER: dict[Property, Callable[[BytesReader], Any]] = { 

170 Property.PAYLOAD_FORMAT_INDICATOR: _read_bool, 

171 Property.MESSAGE_EXPIRY_INTERVAL: BytesReader.read_uint32, 

172 Property.CONTENT_TYPE: BytesReader.read_str, 

173 Property.RESPONSE_TOPIC: BytesReader.read_str, 

174 Property.CORRELATION_DATA: BytesReader.read_binary, 

175 Property.SUBSCRIPTION_IDENTIFIER: BytesReader.read_variable_byte_integer, 

176 Property.SESSION_EXPIRY_INTERVAL: BytesReader.read_uint32, 

177 Property.ASSIGNED_CLIENT_IDENTIFIER: BytesReader.read_str, 

178 Property.SERVER_KEEP_ALIVE: BytesReader.read_uint16, 

179 Property.AUTHENTICATION_METHOD: BytesReader.read_str, 

180 Property.AUTHENTICATION_DATA: BytesReader.read_binary, 

181 Property.REQUEST_PROBLEM_INFORMATION: _read_bool, 

182 Property.WILL_DELAY_INTERVAL: BytesReader.read_uint32, 

183 Property.REQUEST_RESPONSE_INFORMATION: _read_bool, 

184 Property.RESPONSE_INFORMATION: BytesReader.read_str, 

185 Property.SERVER_REFERENCE: BytesReader.read_str, 

186 Property.REASON_STRING: BytesReader.read_str, 

187 Property.RECEIVE_MAXIMUM: BytesReader.read_uint16, 

188 Property.TOPIC_ALIAS_MAXIMUM: BytesReader.read_uint16, 

189 Property.TOPIC_ALIAS: BytesReader.read_uint16, 

190 Property.MAXIMUM_QOS: BytesReader.read_byte, 

191 Property.RETAIN_AVAILABLE: _read_bool, 

192 Property.USER_PROPERTY: _read_user_property, 

193 Property.MAXIMUM_PACKET_SIZE: BytesReader.read_uint32, 

194 Property.WILDCARD_SUBSCRIPTION_AVAILABLE: _read_bool, 

195 Property.SUBSCRIPTION_IDENTIFIER_AVAILABLE: _read_bool, 

196 Property.SHARED_SUBSCRIPTION_AVAILABLE: _read_bool, 

197} 

198 

199 

200_pack_uint8: Callable[[int], bytes] = struct.Struct("!B").pack 

201_pack_uint16: Callable[[int], bytes] = struct.Struct("!H").pack 

202_pack_uint32: Callable[[int], bytes] = struct.Struct("!L").pack 

203 

204_MAP_PROPERTY_PACKERS: dict[Property, Callable[[Any], Union[bytes, bytearray]]] = { 

205 Property.PAYLOAD_FORMAT_INDICATOR: _pack_uint8, 

206 Property.MESSAGE_EXPIRY_INTERVAL: _pack_uint32, 

207 Property.CONTENT_TYPE: pack_str16, 

208 Property.RESPONSE_TOPIC: pack_str16, 

209 Property.CORRELATION_DATA: pack_binary, 

210 Property.SUBSCRIPTION_IDENTIFIER: pack_variable_byte_integer, 

211 Property.SESSION_EXPIRY_INTERVAL: _pack_uint32, 

212 Property.ASSIGNED_CLIENT_IDENTIFIER: pack_str16, 

213 Property.SERVER_KEEP_ALIVE: _pack_uint16, 

214 Property.AUTHENTICATION_METHOD: pack_str16, 

215 Property.AUTHENTICATION_DATA: pack_binary, 

216 Property.REQUEST_PROBLEM_INFORMATION: _pack_uint8, 

217 Property.WILL_DELAY_INTERVAL: _pack_uint32, 

218 Property.REQUEST_RESPONSE_INFORMATION: _pack_uint8, 

219 Property.RESPONSE_INFORMATION: pack_str16, 

220 Property.SERVER_REFERENCE: pack_str16, 

221 Property.REASON_STRING: pack_str16, 

222 Property.RECEIVE_MAXIMUM: _pack_uint16, 

223 Property.TOPIC_ALIAS_MAXIMUM: _pack_uint16, 

224 Property.TOPIC_ALIAS: _pack_uint16, 

225 Property.MAXIMUM_QOS: _pack_uint8, 

226 Property.RETAIN_AVAILABLE: _pack_uint8, 

227 Property.USER_PROPERTY: pack_str16, 

228 Property.MAXIMUM_PACKET_SIZE: _pack_uint32, 

229 Property.WILDCARD_SUBSCRIPTION_AVAILABLE: _pack_uint8, 

230 Property.SUBSCRIPTION_IDENTIFIER_AVAILABLE: _pack_uint8, 

231 Property.SHARED_SUBSCRIPTION_AVAILABLE: _pack_uint8, 

232} 

233 

234 

235# may be in a packet more than once, parsed into a list 

236_REPEATABLE_PROPERTIES: Final = frozenset( 

237 {Property.USER_PROPERTY, Property.SUBSCRIPTION_IDENTIFIER} 

238) 

239 

240 

241def parse_properties(reader: BytesReader) -> Properties: 

242 """Reads the length of properties and the properties.""" 

243 length = reader.read_variable_byte_integer() 

244 

245 if not length: 

246 return {} 

247 

248 if length > reader.remaining(): 248 ↛ 249line 248 didn't jump to line 249 because the condition on line 248 was never true

249 raise MalformedPacketError("Properties are longer than the packet") 

250 

251 end = reader.remaining() - length 

252 # filled by names of the enum, which are the keys of Properties 

253 properties: dict[str, Any] = {} 

254 

255 while reader.remaining() > end: 

256 code = reader.read_variable_byte_integer() 

257 

258 try: 

259 prop = Property(code) 

260 except ValueError as exc: 

261 raise MalformedPacketError(f"Unknown property 0x{code:02X}") from exc 

262 

263 value = _MAP_PROPERTY_PARSER[prop](reader) 

264 

265 if prop in _REPEATABLE_PROPERTIES: 

266 properties.setdefault(prop.name.lower(), []).append(value) 

267 else: 

268 properties[prop.name.lower()] = value 

269 

270 if reader.remaining() != end: 270 ↛ 271line 270 didn't jump to line 271 because the condition on line 270 was never true

271 raise MalformedPacketError("Properties don't match their length") 

272 

273 return cast(Properties, properties) 

274 

275 

276def pack_properties(properties: Properties) -> bytes: 

277 properties = properties or {} 

278 

279 data = bytearray() 

280 

281 for prop_name, prop_value in properties.items(): 

282 code = _NAME_TO_CODE_MAP[prop_name] 

283 value_packer = _MAP_PROPERTY_PACKERS[code] 

284 

285 if code == Property.USER_PROPERTY: 

286 for value in cast(Sequence[Tuple[str, str]], prop_value): 

287 data.append(code & 0xFF) 

288 data += value_packer(value[0]) 

289 data += value_packer(value[1]) 

290 elif code == Property.SUBSCRIPTION_IDENTIFIER and not isinstance( 

291 prop_value, int 

292 ): 

293 for identifier in cast(Sequence[int], prop_value): 

294 data.append(code & 0xFF) 

295 data += value_packer(identifier) 

296 else: 

297 data.append(code & 0xFF) 

298 

299 data.extend(value_packer(prop_value)) 

300 

301 result = pack_variable_byte_integer(len(data)) 

302 result += data 

303 

304 return bytes(result)