Properties & string validation

Summary: related to T13318

Reviewers: ivica

Reviewed By: ivica

Subscribers: miljen, iljazovic

Differential Revision: https://repo.mireo.local/D26975

doc/qbk/reference/Error_handling.qbk

@@ -31,6 +31,11 @@ may complete with, along with the reasons for their occurrence.
 		Therefore, the Client should re-subscribe.
 		This error code is exclusive to completion handlers associated with [refmem mqtt_client async_receive] calls.
 	]]
+	[[`async_mqtt5::client::error::malformed_packet`][
+		The Client has attempted to send a packet that does not conform to the specification.
+		This issue can arise from improperly formed UTF-8 encoded strings.
+		Additionally, this error can be caused by providing out-of-range values.
+	]]
 	[[`async_mqtt5::client::error::pid_overrun`] [
 		This error code signifies that the Client was unable to allocate a Packet Identifier for
 		the current operation due to the exhaustion of the available identifiers.

include/async_mqtt5/detail/utf8_mqtt.hpp

@@ -90,7 +90,7 @@ validation_result validate_impl(
 	return validation_result::valid;
 }
 
-inline validation_result is_valid_mqtt_utf8(std::string_view str) {
+inline validation_result validate_mqtt_utf8(std::string_view str) {
 	return validate_impl(str, is_valid_string_size, is_utf8);
 }
 

include/async_mqtt5/error.hpp

@@ -53,10 +53,8 @@ namespace client {
  * \details Represents error that occur on the client side.
  */
 enum class error : int {
-	/// \cond INTERNAL
+	/** The packet is malformed. */
-	/** Malformed packet has been detected. */
 	malformed_packet = 100,
-	/// \endcond
 
 	/** The Client's session does not exist or it has expired. */
 	session_expired,
@@ -92,7 +90,7 @@ enum class error : int {
 inline std::string client_error_to_string(error err) {
 	switch (err) {
 		case error::malformed_packet:
-			return "Malformed packet has been detected";
+			return "The packet is malformed.";
 		case error::session_expired:
 			return "The Client's session does not exist or it has expired.";
 		case error::pid_overrun:
@@ -101,7 +99,7 @@ inline std::string client_error_to_string(error err) {
 			return "The Topic is invalid and "
 				"does not conform to the specification.";
 		case error::qos_not_supported:
-			return "The Server does not support the specified QoS";
+			return "The Server does not support the specified QoS.";
 		case error::retain_not_available:
 			return "The Server does not support retained messages.";
 		case error::topic_alias_maximum_reached:
@@ -114,7 +112,7 @@ inline std::string client_error_to_string(error err) {
 		case error::shared_subscription_not_available:
 			return "The Server does not support Shared Subscriptions.";
 		default:
-			return "Unknown client error";
+			return "Unknown client error.";
 	}
 }
 

include/async_mqtt5/impl/disconnect_op.hpp

@@ -7,9 +7,10 @@
 
 #include <async_mqtt5/types.hpp>
 
+#include <async_mqtt5/detail/cancellable_handler.hpp>
 #include <async_mqtt5/detail/control_packet.hpp>
 #include <async_mqtt5/detail/internal_types.hpp>
-#include <async_mqtt5/detail/cancellable_handler.hpp>
+#include <async_mqtt5/detail/utf8_mqtt.hpp>
 
 #include <async_mqtt5/impl/codecs/message_encoders.hpp>
 
@@ -60,6 +61,10 @@ public:
 	}
 
 	void perform() {
+		error_code ec = validate_disconnect(_context.props);
+		if (ec)
+			return complete_post(ec);
+
 		auto disconnect = control_packet<allocator_type>::of(
 			no_pid, get_allocator(),
 			encoders::encode_disconnect,
@@ -108,9 +113,28 @@ public:
 	}
 
 private:
+	static error_code validate_disconnect(const disconnect_props& props) {
+		auto reason_string = props[prop::reason_string];
+		if (
+			reason_string &&
+			validate_mqtt_utf8(*reason_string) != validation_result::valid
+		)
+			return client::error::malformed_packet;
+
+		auto user_properties = props[prop::user_property];
+		for (const auto& user_prop: user_properties)
+			if (validate_mqtt_utf8(user_prop) != validation_result::valid)
+				return client::error::malformed_packet;
+		return error_code {};
+	}
+
 	void complete(error_code ec) {
 		_handler.complete(ec);
 	}
+
+	void complete_post(error_code ec) {
+		_handler.complete_post(ec);
+	}
 };
 
 template <typename ClientService, typename CompletionToken>

include/async_mqtt5/impl/publish_send_op.hpp

@@ -11,6 +11,7 @@
 #include <async_mqtt5/detail/control_packet.hpp>
 #include <async_mqtt5/detail/internal_types.hpp>
 #include <async_mqtt5/detail/topic_validation.hpp>
+#include <async_mqtt5/detail/utf8_mqtt.hpp>
 
 #include <async_mqtt5/impl/disconnect_op.hpp>
 #include <async_mqtt5/impl/codecs/message_decoders.hpp>
@@ -99,7 +100,7 @@ public:
 		std::string topic, std::string payload,
 		retain_e retain, const publish_props& props
 	) {
-		auto ec = validate_publish(topic, retain, props);
+		auto ec = validate_publish(topic, payload, retain, props);
 		if (ec)
 			return complete_post(ec);
 
@@ -340,35 +341,78 @@ public:
 
 
 private:
+	error_code validate_props(
+		const publish_props& props,
+		prop::value_type_t<prop::topic_alias_maximum> topic_alias_max_opt
+	) {
+		auto topic_alias = props[prop::topic_alias];
+		if (topic_alias) {
+			auto topic_alias_max = topic_alias_max_opt.value_or(0);
+
+			if (topic_alias_max == 0 || *topic_alias > topic_alias_max)
+				return client::error::topic_alias_maximum_reached;
+			if (*topic_alias == 0 )
+				return client::error::malformed_packet;
+		}
+
+		auto response_topic = props[prop::response_topic];
+		if (
+			response_topic &&
+			validate_topic_name(*response_topic) != validation_result::valid
+		)
+			return client::error::malformed_packet;
+
+		auto user_properties = props[prop::user_property];
+		for (const auto& user_prop: user_properties)
+			if (validate_mqtt_utf8(user_prop) != validation_result::valid)
+				return client::error::malformed_packet;
+
+		auto subscription_identifier = props[prop::subscription_identifier];
+		if (
+			subscription_identifier &&
+			(*subscription_identifier < 1 || *subscription_identifier > 268'435'455)
+		)
+			return client::error::malformed_packet;
+
+		auto content_type = props[prop::content_type];
+		if (
+			content_type &&
+			validate_mqtt_utf8(*content_type) != validation_result::valid
+		)
+			return client::error::malformed_packet;
+
+		return error_code {};
+	}
+
 	error_code validate_publish(
-		const std::string& topic, retain_e retain, const publish_props& props
+		const std::string& topic, const std::string& payload,
+		retain_e retain, const publish_props& props
 	) {
 		if (validate_topic_name(topic) != validation_result::valid)
 			return client::error::invalid_topic;
 
-		const auto& [max_qos, retain_avail, topic_alias_max] =
+		const auto& [max_qos_opt, retain_available_opt, topic_alias_max_opt] =
 			_svc_ptr->connack_props(
 				prop::maximum_qos, prop::retain_available,
 				prop::topic_alias_maximum
 			);
+		auto max_qos = max_qos_opt.value_or(2);
+		auto retain_available = retain_available_opt.value_or(1);
 
-		if (max_qos && uint8_t(qos_type) > *max_qos)
+		if (uint8_t(qos_type) > max_qos)
 			return client::error::qos_not_supported;
 
-		if (retain_avail && *retain_avail == 0 && retain == retain_e::yes)
+		if (retain_available == 0 && retain == retain_e::yes)
 			return client::error::retain_not_available;
 
-		auto topic_alias = props[prop::topic_alias];
+		auto payload_format = props[prop::payload_format_indicator].value_or(0);
 		if (
-			(!topic_alias_max || topic_alias_max && *topic_alias_max == 0) &&
+			payload_format == 1 &&
-			topic_alias
+			validate_mqtt_utf8(payload) != validation_result::valid
 		)
-			return client::error::topic_alias_maximum_reached;
+			return client::error::malformed_packet;
 
-		if (topic_alias_max && topic_alias && *topic_alias > *topic_alias_max)
+		return validate_props(props, topic_alias_max_opt);
-			return client::error::topic_alias_maximum_reached;
-
-		return error_code {};
 	}
 
 	void on_malformed_packet(const std::string& reason) {

include/async_mqtt5/impl/subscribe_op.hpp

@@ -156,6 +156,11 @@ private:
 	static error_code validate_props(
 		const subscribe_props& props, bool sub_id_available
 	) {
+		auto user_properties = props[prop::user_property];
+		for (const auto& user_prop: user_properties)
+			if (validate_mqtt_utf8(user_prop) != validation_result::valid)
+				return client::error::malformed_packet;
+
 		auto sub_id = props[prop::subscription_identifier];
 		if (!sub_id.has_value())
 			return error_code {};

include/async_mqtt5/impl/unsubscribe_op.hpp

@@ -59,7 +59,7 @@ public:
 		const std::vector<std::string>& topics,
 		const unsubscribe_props& props
 	) {
-		auto ec = validate_topics(topics);
+		auto ec = validate_unsubscribe(topics, props);
 		if (ec)
 			return complete_post(ec, topics.size());
 
@@ -147,10 +147,18 @@ public:
 
 private:
 
-	static error_code validate_topics(const std::vector<std::string>& topics) {
+	static error_code validate_unsubscribe(
+		const std::vector<std::string>& topics,
+		const unsubscribe_props& props
+	) {
 		for (const auto& topic : topics)
 			if (validate_topic_filter(topic) != validation_result::valid)
 				return client::error::invalid_topic;
+
+		auto user_properties = props[prop::user_property];
+		for (const auto& user_prop: user_properties)
+			if (validate_mqtt_utf8(user_prop) != validation_result::valid)
+				return client::error::malformed_packet;
 		return error_code {};
 	}
 

include/async_mqtt5/mqtt_client.hpp

@@ -361,6 +361,7 @@ public:
 	 *		- `boost::system::errc::errc_t::success` \n
 	 *		- `boost::asio::error::operation_aborted` \n
 	 *		- `boost::asio::error::no_recovery` \n
+	 *		- \link async_mqtt5::client::error::malformed_packet \endlink
 	 *		- \link async_mqtt5::client::error::pid_overrun \endlink
 	 *		- \link async_mqtt5::client::error::qos_not_supported \endlink
 	 *		- \link async_mqtt5::client::error::retain_not_available \endlink
@@ -436,6 +437,7 @@ public:
 	 *		- `boost::system::errc::errc_t::success` \n
 	 *		- `boost::asio::error::no_recovery` \n
 	 *		- `boost::asio::error::operation_aborted` \n
+	 *		- \link async_mqtt5::client::error::malformed_packet \endlink
 	 *		- \link async_mqtt5::client::error::pid_overrun \endlink
 	 *		- \link async_mqtt5::client::error::invalid_topic \endlink
 	 *		- \link async_mqtt5::client::error::wildcard_subscription_not_available \endlink
@@ -506,6 +508,7 @@ public:
 	 *		- `boost::system::errc::errc_t::success` \n
 	 *		- `boost::asio::error::no_recovery` \n
 	 *		- `boost::asio::error::operation_aborted` \n
+	 *		- \link async_mqtt5::client::error::malformed_packet \endlink
 	 *		- \link async_mqtt5::client::error::pid_overrun \endlink
 	 *		- \link async_mqtt5::client::error::invalid_topic \endlink
 	 *		- \link async_mqtt5::client::error::wildcard_subscription_not_available \endlink
@@ -564,6 +567,7 @@ public:
 	 *		- `boost::system::errc::errc_t::success` \n
 	 *		- `boost::asio::error::no_recovery` \n
 	 *		- `boost::asio::error::operation_aborted` \n
+	 *		- \link async_mqtt5::client::error::malformed_packet \endlink
 	 *		- \link async_mqtt5::client::error::pid_overrun \endlink
 	 *		- \link async_mqtt5::client::error::invalid_topic \endlink
 	 *
@@ -630,6 +634,7 @@ public:
 	 *		- `boost::system::errc::errc_t::success` \n
 	 *		- `boost::asio::error::no_recovery` \n
 	 *		- `boost::asio::error::operation_aborted` \n
+	 *		- \link async_mqtt5::client::error::malformed_packet \endlink
 	 *		- \link async_mqtt5::client::error::pid_overrun \endlink
 	 *		- \link async_mqtt5::client::error::invalid_topic \endlink
 	 *
@@ -737,6 +742,7 @@ public:
 	 *	The list of all possible error codes that this operation can finish with:\n
 	 *		- `boost::system::errc::errc_t::success`\n
 	 *		- `boost::asio::error::operation_aborted`\n
+	 *		- \link async_mqtt5::client::error::malformed_packet \endlink
 	 *
 	 * Refer to the section on \__ERROR_HANDLING\__ to find the underlying causes for each error code.
 	 */

test/unit/test/publish_send_op.cpp

@@ -84,6 +84,152 @@ BOOST_AUTO_TEST_CASE(test_invalid_topic_names) {
 	BOOST_CHECK_EQUAL(handlers_called, expected_handlers_called);
 }
 
+BOOST_AUTO_TEST_CASE(test_malformed_packet) {
+	std::string malformed_str = std::string { 0x01 };
+
+	publish_props malfored_response_topic_props;
+	malfored_response_topic_props[prop::response_topic] = "response#topic";
+
+	publish_props utf8_payload_props;
+	utf8_payload_props[prop::payload_format_indicator] = uint8_t(1);
+
+	publish_props invalid_user_props;
+	invalid_user_props[prop::user_property].push_back(malformed_str);
+
+	publish_props malformed_content_type_props;
+	malformed_content_type_props[prop::content_type] = malformed_str;
+
+	publish_props out_of_range_subid_props;
+	out_of_range_subid_props[prop::subscription_identifier] = 300'000'000;
+
+	std::vector<publish_props> testing_props = {
+		malfored_response_topic_props, utf8_payload_props,
+		invalid_user_props, malformed_content_type_props,
+		out_of_range_subid_props
+	};
+
+	int expected_handlers_called = testing_props.size();
+	int handlers_called = 0;
+
+	asio::io_context ioc;
+	using client_service_type = test::test_service<asio::ip::tcp::socket>;
+	auto svc_ptr = std::make_shared<client_service_type>(ioc.get_executor());
+
+	for (const auto& props: testing_props) {
+		auto handler = [&handlers_called](error_code ec) {
+			++handlers_called;
+			BOOST_CHECK(ec == client::error::malformed_packet);
+		};
+
+		detail::publish_send_op<
+			client_service_type, decltype(handler), qos_e::at_most_once
+		> { svc_ptr, std::move(handler) }
+		.perform("topic", malformed_str, retain_e::no, props);
+	}
+
+	ioc.run();
+	BOOST_CHECK_EQUAL(handlers_called, expected_handlers_called);
+}
+
+BOOST_AUTO_TEST_CASE(test_qos_not_supported) {
+	connack_props props;
+	props[prop::maximum_qos] = uint8_t(0);
+
+	constexpr int expected_handlers_called = 1;
+	int handlers_called = 0;
+
+	asio::io_context ioc;
+	using client_service_type = test::test_service<asio::ip::tcp::socket>;
+	auto svc_ptr = std::make_shared<client_service_type>(
+		ioc.get_executor(), std::move(props)
+	);
+
+	auto handler = [&handlers_called](error_code ec, reason_code rc, puback_props) {
+		++handlers_called;
+		BOOST_CHECK(ec == client::error::qos_not_supported);
+		BOOST_CHECK_EQUAL(rc, reason_codes::empty);
+	};
+
+	detail::publish_send_op<
+		client_service_type, decltype(handler), qos_e::at_least_once
+	> { svc_ptr, std::move(handler) }
+	.perform(
+		"test", "payload", retain_e::no, {}
+	);
+
+	ioc.run();
+	BOOST_CHECK_EQUAL(handlers_called, expected_handlers_called);
+}
+
+BOOST_AUTO_TEST_CASE(test_retain_not_available) {
+	connack_props props;
+	props[prop::retain_available] = uint8_t(0);
+
+	constexpr int expected_handlers_called = 1;
+	int handlers_called = 0;
+
+	asio::io_context ioc;
+	using client_service_type = test::test_service<asio::ip::tcp::socket>;
+	auto svc_ptr = std::make_shared<client_service_type>(
+		ioc.get_executor(), std::move(props)
+	);
+
+	auto handler = [&handlers_called](error_code ec, reason_code rc, puback_props) {
+		++handlers_called;
+		BOOST_CHECK(ec == client::error::retain_not_available);
+		BOOST_CHECK_EQUAL(rc, reason_codes::empty);
+	};
+
+	detail::publish_send_op<
+		client_service_type, decltype(handler), qos_e::at_least_once
+	> { svc_ptr, std::move(handler) }
+	.perform(
+		"test", "payload", retain_e::yes, {}
+	);
+
+	ioc.run();
+	BOOST_CHECK_EQUAL(handlers_called, expected_handlers_called);
+}
+
+BOOST_AUTO_TEST_CASE(test_topic_alias_maximum) {
+	connack_props ta_allowed_props;
+	ta_allowed_props[prop::topic_alias_maximum] = uint16_t(10);
+
+	std::vector<connack_props> test_props = {
+		ta_allowed_props, connack_props {} /* not allowed */
+	};
+
+	int expected_handlers_called = test_props.size();
+	int handlers_called = 0;
+
+	asio::io_context ioc;
+
+	for (const auto& ca_props: test_props) {
+		using client_service_type = test::test_service<asio::ip::tcp::socket>;
+		auto svc_ptr = std::make_shared<client_service_type>(
+			ioc.get_executor(), ca_props
+		);
+
+		auto handler = [&handlers_called](error_code ec) {
+			++handlers_called;
+			BOOST_CHECK(ec == client::error::topic_alias_maximum_reached);
+		};
+
+		publish_props props;
+		props[prop::topic_alias] = uint16_t(12);
+
+		detail::publish_send_op<
+			client_service_type, decltype(handler), qos_e::at_most_once
+		> { svc_ptr, std::move(handler) }
+		.perform(
+			"test", "payload", retain_e::yes, props
+		);
+	}
+
+	ioc.run();
+	BOOST_CHECK_EQUAL(handlers_called, expected_handlers_called);
+}
+
 BOOST_AUTO_TEST_CASE(test_publish_immediate_cancellation) {
 	constexpr int expected_handlers_called = 1;
 	int handlers_called = 0;

test/unit/test/string_validation.cpp

@@ -32,24 +32,24 @@ std::string to_str(int utf8ch) {
 BOOST_AUTO_TEST_CASE(utf8_string_validation) {
 	using namespace async_mqtt5::detail;
 
-	BOOST_CHECK(is_valid_mqtt_utf8("stringy") == validation_result::valid);
+	BOOST_CHECK(validate_mqtt_utf8("stringy") == validation_result::valid);
-	BOOST_CHECK(is_valid_mqtt_utf8("") == validation_result::valid);
+	BOOST_CHECK(validate_mqtt_utf8("") == validation_result::valid);
-	BOOST_CHECK(is_valid_mqtt_utf8(std::string(75000, 'a')) == validation_result::invalid);
+	BOOST_CHECK(validate_mqtt_utf8(std::string(75000, 'a')) == validation_result::invalid);
 
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0x1)) == validation_result::invalid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0x1)) == validation_result::invalid);
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0x1F)) == validation_result::invalid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0x1F)) == validation_result::invalid);
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0x20)) == validation_result::valid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0x20)) == validation_result::valid);
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0x7E)) == validation_result::valid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0x7E)) == validation_result::valid);
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0x7F)) == validation_result::invalid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0x7F)) == validation_result::invalid);
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0x9F)) == validation_result::invalid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0x9F)) == validation_result::invalid);
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0xA0)) == validation_result::valid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0xA0)) == validation_result::valid);
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0xD800)) == validation_result::invalid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0xD800)) == validation_result::invalid);
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0xDFFF)) == validation_result::invalid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0xDFFF)) == validation_result::invalid);
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0xFDD0)) == validation_result::invalid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0xFDD0)) == validation_result::invalid);
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0xFDEF)) == validation_result::invalid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0xFDEF)) == validation_result::invalid);
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0xFDF0)) == validation_result::valid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0xFDF0)) == validation_result::valid);
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0x1FFFE)) == validation_result::invalid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0x1FFFE)) == validation_result::invalid);
-	BOOST_CHECK(is_valid_mqtt_utf8(to_str(0x1FFFF)) == validation_result::invalid);
+	BOOST_CHECK(validate_mqtt_utf8(to_str(0x1FFFF)) == validation_result::invalid);
 }
 
 BOOST_AUTO_TEST_CASE(topic_filter_validation) {

test/unit/test/subscribe_op.cpp

@@ -141,7 +141,6 @@ BOOST_AUTO_TEST_CASE(test_large_subscription_id) {
 BOOST_AUTO_TEST_CASE(test_subscription_ids_not_supported) {
 	connack_props props;
 	props[prop::subscription_identifier_available] = uint8_t(0);
-	BOOST_ASSERT(props[prop::subscription_identifier_available] == 0);
 
 	constexpr int expected_handlers_called = 1;
 	int handlers_called = 0;