Naive reading of MQTT 5 properties. No processing done.

lib/packet_mosq.c

@@ -280,16 +280,18 @@ void packet__write_uint32(struct mosquitto__packet *packet, uint32_t word)
 }
 
 
-int packet__read_varint(struct mosquitto__packet *packet, uint32_t *word)
+int packet__read_varint(struct mosquitto__packet *packet, int32_t *word, uint8_t *bytes)
 {
 	int i;
 	int remaining_mult = 1;
 	uint8_t byte;
 
 	*word = 0;
+	if(bytes) (*bytes) = 0;
 
 	for(i=0; i<4; i++){
 		if(packet->pos < packet->remaining_length){
+			if(bytes) (*bytes)++;
 			byte = packet->payload[packet->pos];
 			*word += (byte & 127) * remaining_mult;
 			remaining_mult *= 128;

lib/packet_mosq.h

@@ -32,7 +32,7 @@ int packet__read_bytes(struct mosquitto__packet *packet, void *bytes, uint32_t c
 int packet__read_string(struct mosquitto__packet *packet, char **str, int *length);
 int packet__read_uint16(struct mosquitto__packet *packet, uint16_t *word);
 int packet__read_uint32(struct mosquitto__packet *packet, uint32_t *word);
-int packet__read_varint(struct mosquitto__packet *packet, uint32_t *word);
+int packet__read_varint(struct mosquitto__packet *packet, int32_t *word, uint8_t *bytes);
 
 void packet__write_byte(struct mosquitto__packet *packet, uint8_t byte);
 void packet__write_bytes(struct mosquitto__packet *packet, const void *bytes, uint32_t count);

lib/property_mosq.c

@@ -0,0 +1,261 @@
+/*
+Copyright (c) 2018 Roger Light <roger@atchoo.org>
+
+All rights reserved. This program and the accompanying materials
+are made available under the terms of the Eclipse Public License v1.0
+and Eclipse Distribution License v1.0 which accompany this distribution.
+
+The Eclipse Public License is available at
+   http://www.eclipse.org/legal/epl-v10.html
+and the Eclipse Distribution License is available at
+  http://www.eclipse.org/org/documents/edl-v10.php.
+
+Contributors:
+   Roger Light - initial implementation and documentation.
+*/
+
+#include "config.h"
+
+#include <assert.h>
+#include <errno.h>
+#include <string.h>
+
+#include "logging_mosq.h"
+#include "memory_mosq.h"
+#include "mqtt_protocol.h"
+#include "packet_mosq.h"
+#include "property_mosq.h"
+
+int property__read(struct mosquitto__packet *packet, uint32_t *len)
+{
+	int rc;
+	uint8_t byte;
+	uint16_t int16;
+	uint32_t int32;
+	char *str;
+	int slen;
+	*len -= 14;
+
+	rc = packet__read_byte(packet, &byte);
+	if(rc) return rc;
+	*len -= 1;
+
+	switch(byte){
+		case PROP_PAYLOAD_FORMAT_INDICATOR:
+			rc = packet__read_byte(packet, &byte);
+			if(rc) return rc;
+			*len -= 1; /* byte */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Payload format indicator: %d", byte);
+			break;
+
+		case PROP_MESSAGE_EXPIRY_INTERVAL:
+			rc = packet__read_uint32(packet, &int32);
+			if(rc) return rc;
+			*len -= 4; /* uint32 */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Message expiry: %d", int32);
+			break;
+
+		case PROP_CONTENT_TYPE:
+			rc = packet__read_string(packet, &str, &slen);
+			if(rc) return rc;
+			*len -= 2 - slen; /* int16, string len */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Content type: %s", str);
+			break;
+
+		case PROP_RESPONSE_TOPIC:
+			rc = packet__read_string(packet, &str, &slen);
+			if(rc) return rc;
+			*len -= 2 - slen; /* int16, string len */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Response topic: %s", str);
+			break;
+
+		case PROP_CORRELATION_DATA:
+			rc = packet__read_string(packet, &str, &slen);
+			if(rc) return rc;
+			*len -= 2 - slen; /* int16, string len */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Correlation data: %s", str);
+			break;
+
+		case PROP_SUBSCRIPTION_IDENTIFIER:
+			rc = packet__read_varint(packet, &int32, &byte);
+			*len -= byte;
+			if(rc) return rc;
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Subscription identifier: %d", int32);
+			break;
+
+		case PROP_SESSION_EXPIRY_INTERVAL:
+			rc = packet__read_uint32(packet, &int32);
+			if(rc) return rc;
+			*len -= 4; /* uint32 */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Session expiry: %d", int32);
+			break;
+
+		case PROP_ASSIGNED_CLIENT_IDENTIFIER:
+			rc = packet__read_string(packet, &str, &slen);
+			if(rc) return rc;
+			*len -= 2 - slen; /* int16, string len */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Assigned client identifier: %s", str);
+			break;
+
+		case PROP_SERVER_KEEP_ALIVE:
+			rc = packet__read_uint16(packet, &int16);
+			if(rc) return rc;
+			*len -= 2; /* uint16 */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Session expiry: %d", int16);
+			break;
+
+		case PROP_AUTHENTICATION_METHOD:
+			rc = packet__read_string(packet, &str, &slen);
+			if(rc) return rc;
+			*len -= 2 - slen; /* int16, string len */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Authentication method: %s", str);
+			break;
+
+		case PROP_AUTHENTICATION_DATA:
+			rc = packet__read_string(packet, &str, &slen);
+			if(rc) return rc;
+			*len -= 2 - slen; /* int16, string len */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Authentication data: %s", str);
+			break;
+
+		case PROP_REQUEST_PROBLEM_INFO:
+			rc = packet__read_byte(packet, &byte);
+			if(rc) return rc;
+			*len -= 1; /* byte */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Request problem information: %d", byte);
+			break;
+
+		case PROP_WILL_DELAY_INTERVAL:
+			rc = packet__read_uint32(packet, &int32);
+			if(rc) return rc;
+			*len -= 4; /* uint32 */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Will delay interval: %d", int32);
+			break;
+
+		case PROP_REQUEST_RESPONSE_INFO:
+			rc = packet__read_byte(packet, &byte);
+			if(rc) return rc;
+			*len -= 1; /* byte */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Request response information: %d", byte);
+			break;
+
+		case PROP_RESPONSE_INFO:
+			rc = packet__read_string(packet, &str, &slen);
+			if(rc) return rc;
+			*len -= 2 - slen; /* int16, string len */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Response information: %s", str);
+			break;
+
+		case PROP_SERVER_REFERENCE:
+			rc = packet__read_string(packet, &str, &slen);
+			if(rc) return rc;
+			*len -= 2 - slen; /* int16, string len */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Server reference: %s", str);
+			break;
+
+		case PROP_REASON_STRING:
+			rc = packet__read_string(packet, &str, &slen);
+			if(rc) return rc;
+			*len -= 2 - slen; /* int16, string len */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Reason string: %s", str);
+			break;
+
+		case PROP_RECEIVE_MAXIMUM:
+			rc = packet__read_uint16(packet, &int16);
+			if(rc) return rc;
+			*len -= 2; /* uint16 */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Receive maximum: %d", int16);
+			break;
+
+		case PROP_TOPIC_ALIAS_MAXIMUM:
+			rc = packet__read_uint16(packet, &int16);
+			if(rc) return rc;
+			*len -= 2; /* uint16 */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Topic alias maximum: %d", int16);
+			break;
+
+		case PROP_TOPIC_ALIAS:
+			rc = packet__read_uint16(packet, &int16);
+			if(rc) return rc;
+			*len -= 2; /* uint16 */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Topic alias: %d", int16);
+			break;
+
+		case PROP_MAXIMUM_QOS:
+			rc = packet__read_byte(packet, &byte);
+			if(rc) return rc;
+			*len -= 1; /* byte */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Maximum QoS: %d", byte);
+			break;
+
+		case PROP_RETAIN_AVAILABLE:
+			rc = packet__read_byte(packet, &byte);
+			if(rc) return rc;
+			*len -= 1; /* byte */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Retain available: %d", byte);
+			break;
+
+		case PROP_USER_PROPERTY:
+			rc = packet__read_string(packet, &str, &slen);
+			if(rc) return rc;
+			*len -= 2 - slen; /* int16, string len */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "User property name: %s", str);
+
+			rc = packet__read_string(packet, &str, &slen);
+			if(rc) return rc;
+			*len -= 2 - slen; /* int16, string len */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "User property value: %s", str);
+			break;
+
+		case PROP_MAXIMUM_PACKET_SIZE:
+			rc = packet__read_uint32(packet, &int32);
+			if(rc) return rc;
+			*len -= 4; /* uint32 */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Maximum packet size: %d", int32);
+			break;
+
+		case PROP_WILDCARD_SUB_AVAILABLE:
+			rc = packet__read_byte(packet, &byte);
+			if(rc) return rc;
+			*len -= 1; /* byte */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Wildcard subscription available: %d", byte);
+			break;
+
+		case PROP_SUBSCRIPTION_ID_AVAILABLE:
+			rc = packet__read_byte(packet, &byte);
+			if(rc) return rc;
+			*len -= 1; /* byte */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Subscription identifier available: %d", byte);
+			break;
+
+		case PROP_SHARED_SUB_AVAILABLE:
+			rc = packet__read_byte(packet, &byte);
+			if(rc) return rc;
+			*len -= 1; /* byte */
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Shared subscription available: %d", byte);
+			break;
+
+		default:
+			log__printf(NULL, MOSQ_LOG_DEBUG, "Unsupported property type: %d", byte);
+			return 1;
+	}
+
+	return MOSQ_ERR_SUCCESS;
+}
+
+
+int property__read_all(struct mosquitto__packet *packet)
+{
+	int rc;
+	int32_t proplen;
+
+	rc = packet__read_varint(packet, &proplen, NULL);
+	if(rc) return rc;
+
+	while(proplen > 0){
+		rc = property__read(packet, &proplen);
+		if(rc) return rc;
+	}
+
+	return MOSQ_ERR_SUCCESS;
+}

lib/property_mosq.h

@@ -0,0 +1,24 @@
+/*
+Copyright (c) 2018 Roger Light <roger@atchoo.org>
+
+All rights reserved. This program and the accompanying materials
+are made available under the terms of the Eclipse Public License v1.0
+and Eclipse Distribution License v1.0 which accompany this distribution.
+ 
+The Eclipse Public License is available at
+   http://www.eclipse.org/legal/epl-v10.html
+and the Eclipse Distribution License is available at
+  http://www.eclipse.org/org/documents/edl-v10.php.
+ 
+Contributors:
+   Roger Light - initial implementation and documentation.
+*/
+#ifndef PROPERTY_MOSQ_H
+#define PROPERTY_MOSQ_H
+
+#include "mosquitto_internal.h"
+#include "mosquitto.h"
+
+int property__read_all(struct mosquitto__packet *packet);
+
+#endif

src/CMakeLists.txt

@@ -29,6 +29,7 @@ set (MOSQ_SRCS
 	../lib/packet_mosq.c ../lib/packet_mosq.h
 	persist.c persist.h
 	plugin.c
+	../lib/property_mosq.c ../lib/property_mosq.h
 	read_handle.c
 	../lib/read_handle.h
 	subs.c

src/Makefile

@@ -31,6 +31,7 @@ OBJS=	mosquitto.o \
 		net.o \
 		net_mosq.o \
 		packet_mosq.o \
+		property_mosq.o \
 		persist.o \
 		plugin.o \
 		read_handle.o \
@@ -130,6 +131,9 @@ persist.o : persist.c persist.h mosquitto_broker_internal.h
 packet_mosq.o : ../lib/packet_mosq.c ../lib/packet_mosq.h
 	${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@
 
+property_mosq.o : ../lib/property_mosq.c ../lib/property_mosq.h
+	${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@
+
 plugin.o : plugin.c mosquitto_plugin.h mosquitto_broker_internal.h
 	${CROSS_COMPILE}${CC} $(BROKER_CFLAGS) -c $< -o $@
 

src/handle_connect.c

@@ -23,6 +23,7 @@ Contributors:
 #include "mqtt_protocol.h"
 #include "memory_mosq.h"
 #include "packet_mosq.h"
+#include "property_mosq.h"
 #include "send_mosq.h"
 #include "sys_tree.h"
 #include "time_mosq.h"
@@ -240,17 +241,8 @@ int handle__connect(struct mosquitto_db *db, struct mosquitto *context)
 	}
 
 	if(protocol_version == PROTOCOL_VERSION_v5){
-		/* FIXME */
-		uint32_t property_len;
-		if(packet__read_varint(&context->in_packet, &property_len)){
-			rc = 1;
-			goto handle_connect_error;
-		}
-		if(property_len != 0){
-			/* FIXME Temporary fudge because of no property support */
-			rc = 1;
-			goto handle_connect_error;
-		}
+		rc = property__read_all(&context->in_packet);
+		if(rc) return rc;
 	}
 
 	if(packet__read_string(&context->in_packet, &client_id, &slen)){

src/handle_publish.c

@@ -24,6 +24,7 @@ Contributors:
 #include "mqtt_protocol.h"
 #include "memory_mosq.h"
 #include "packet_mosq.h"
+#include "property_mosq.h"
 #include "read_handle.h"
 #include "send_mosq.h"
 #include "sys_tree.h"
@@ -135,6 +136,11 @@ int handle__publish(struct mosquitto_db *db, struct mosquitto *context)
 		}
 	}
 
+	if(context->protocol == mosq_p_mqtt5){
+		rc = property__read_all(&context->in_packet);
+		if(rc) return rc;
+	}
+
 	payloadlen = context->in_packet.remaining_length - context->in_packet.pos;
 	G_PUB_BYTES_RECEIVED_INC(payloadlen);
 	if(context->listener && context->listener->mount_point){