[mqtt-client] add support for re-authentication

Summary:
https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901257

Resolves T12899

Reviewers: ivica

Reviewed By: ivica

Subscribers: korina

Maniphest Tasks: T12899

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

doc/qbk/reference/concepts/is_authenticator.qbk

@@ -7,19 +7,27 @@
 
 [section:is_authenticator is_authenticator concept]
 
-`is_authenticator` represents authenticator object that needs to have following functions:
+A type `Authenticator` satisfies `is_authenticator` concept if it satisifes the requirements listed below.
 
- void async_auth(
-	 ``[reflink2 auth_step_e async_mqtt5::auth_step_e]`` auth_step, // authentication stage
-	 std::string server_data, // server authentication data
-	 ``[asioreflink any_completion_handler any_completion_handler]``<
-		 void(
-			 __ERROR_CODE__ ec, // non-trivial error code aborts authentication
-			 std::string auth_data // client authentication data
-		 )
-	 >
- );
- std::string_view method(); // returns authentication method
+[table
+	[[operation] [type] [arguments]]
+	[
+		[```a.async_auth(step, data, h)```]
+		[`void`]
+		[
+[*`step`] is [reflink2 auth_step_e async_mqtt5::auth_step_e] that specifies current authentication stage.
+
+[*`data`] is `std::string`, server's authentication data.
+
+[*`h`] is [asioreflink any_completion_handler any_completion_handler] with signature `void(__ERROR_CODE__ ec, std::string client_data)`. If `ec` is non-trivial, authentication is aborted.
+		]
+	]
+	[
+		[```a.method()```]
+		[`std::string_view`, authentication method]
+		[]
+	]
+]
 
 
 [endsect]

doc/reference.dox

@@ -71,7 +71,7 @@ WARN_LOGFILE           =
 #---------------------------------------------------------------------------
 INPUT                  = ../include/async_mqtt5/error.hpp \
                          ../include/async_mqtt5/types.hpp \
-                         ../include/async_mqtt5/mqtt_client.hpp \
+                         ../include/async_mqtt5/mqtt_client.hpp
 FILE_PATTERNS          = 
 RECURSIVE              = NO
 EXCLUDE                =

include/async_mqtt5/impl/client_service.hpp

@@ -133,6 +133,9 @@ private:
 	template <typename ClientService>
 	friend class sentry_op;
 
+	template <typename ClientService>
+	friend class re_auth_op;
+
 	stream_context_type _stream_context;
 	stream_type _stream;
 

include/async_mqtt5/impl/internal/codecs/message_decoders.hpp

@@ -147,7 +147,9 @@ using puback_message = std::tuple<
 inline std::optional<puback_message> decode_puback(
 	uint32_t remain_length, byte_citer& it
 ) {
-	auto puback_ =  basic::scope_limit_(remain_length)[
+	if (remain_length == 0)
+		return puback_message {};
+	auto puback_ = basic::scope_limit_(remain_length)[
 		x3::byte_ >> prop::props_<puback_props>
 	];
 	return type_parse(it, it + remain_length, puback_);
@@ -161,7 +163,9 @@ using pubrec_message = std::tuple<
 inline std::optional<pubrec_message> decode_pubrec(
 	uint32_t remain_length, byte_citer& it
 ) {
-	auto pubrec_ =  basic::scope_limit_(remain_length)[
+	if (remain_length == 0)
+		return pubrec_message {};
+	auto pubrec_ = basic::scope_limit_(remain_length)[
 		x3::byte_ >> prop::props_<pubrec_props>
 	];
 	return type_parse(it, it + remain_length, pubrec_);
@@ -175,7 +179,9 @@ using pubrel_message = std::tuple<
 inline std::optional<pubrel_message> decode_pubrel(
 	uint32_t remain_length, byte_citer& it
 ) {
-	auto pubrel_ =  basic::scope_limit_(remain_length)[
+	if (remain_length == 0)
+		return pubrel_message {};
+	auto pubrel_ = basic::scope_limit_(remain_length)[
 		x3::byte_ >> prop::props_<pubrel_props>
 	];
 	return type_parse(it, it + remain_length, pubrel_);
@@ -189,7 +195,9 @@ using pubcomp_message = std::tuple<
 inline std::optional<pubcomp_message> decode_pubcomp(
 	uint32_t remain_length, byte_citer& it
 ) {
-	auto pubcomp_ =  basic::scope_limit_(remain_length)[
+	if (remain_length == 0)
+		return pubcomp_message {};
+	auto pubcomp_ = basic::scope_limit_(remain_length)[
 		x3::byte_ >> prop::props_<pubcomp_props>
 	];
 	return type_parse(it, it + remain_length, pubcomp_);
@@ -259,6 +267,8 @@ using disconnect_message = std::tuple<
 inline std::optional<disconnect_message> decode_disconnect(
 	uint32_t remain_length, byte_citer& it
 ) {
+	if (remain_length == 0)
+		return disconnect_message {};
 	auto disconnect_ = basic::scope_limit_(remain_length)[
 		x3::byte_ >> prop::props_<disconnect_props>
 	];
@@ -273,6 +283,8 @@ using auth_message = std::tuple<
 inline std::optional<auth_message> decode_auth(
 	uint32_t remain_length, byte_citer& it
 ) {
+	if (remain_length == 0)
+		return auth_message {};
 	auto auth_ = basic::scope_limit_(remain_length)[
 		x3::byte_ >> prop::props_<auth_props>
 	];

include/async_mqtt5/impl/re_auth_op.hpp

@@ -0,0 +1,135 @@
+#ifndef ASYNC_MQTT5_RE_AUTH_OP_hpp
+#define ASYNC_MQTT5_RE_AUTH_OP_hpp
+
+#include <boost/asio/detached.hpp>
+
+#include <async_mqtt5/error.hpp>
+
+#include <async_mqtt5/detail/control_packet.hpp>
+
+#include <async_mqtt5/impl/internal/codecs/message_decoders.hpp>
+#include <async_mqtt5/impl/internal/codecs/message_encoders.hpp>
+
+#include <async_mqtt5/impl/disconnect_op.hpp>
+
+namespace async_mqtt5::detail {
+
+namespace asio = boost::asio;
+
+template <typename ClientService>
+class re_auth_op {
+	using client_service = ClientService;
+	struct on_auth_data {};
+
+	std::shared_ptr<client_service> _svc_ptr;
+	any_authenticator& _auth;
+
+public:
+	re_auth_op(
+		const std::shared_ptr<client_service>& svc_ptr
+	) :
+		_svc_ptr(svc_ptr),
+		_auth(_svc_ptr->_stream_context.mqtt_context().authenticator)
+	{}
+
+	re_auth_op(re_auth_op&&) noexcept = default;
+	re_auth_op(const re_auth_op&) noexcept = delete;
+
+	using executor_type = typename client_service::executor_type;
+	executor_type get_executor() const noexcept {
+		return _svc_ptr->get_executor();
+	}
+
+	using allocator_type = asio::recycling_allocator<void>;
+	allocator_type get_allocator() const noexcept {
+		return allocator_type {};
+	}
+
+	void perform() {
+		auto auth_step = auth_step_e::client_initial;
+		return _auth.async_auth(
+			auth_step, "",
+			asio::prepend(std::move(*this), on_auth_data {}, auth_step)
+		);
+	}
+
+	void perform(decoders::auth_message auth_message) {
+		if (_auth.method().empty())
+			return on_auth_fail(
+				"Unexpected AUTH received.",
+				disconnect_rc_e::protocol_error
+			);
+
+		const auto& [rc, auth_props] = auth_message;
+		auto auth_rc = to_reason_code<reason_codes::category::auth>(rc);
+		if (!auth_rc.has_value())
+			return on_auth_fail(
+				"Malformed AUTH received: bad reason code",
+				disconnect_rc_e::malformed_packet
+			);
+
+		auto server_auth_method = auth_props[prop::authentication_method];
+		if (!server_auth_method || *server_auth_method != _auth.method())
+			return on_auth_fail(
+				"Malformed AUTH received: wrong authentication method",
+				disconnect_rc_e::protocol_error
+			);
+
+		auto auth_step = auth_rc == reason_codes::success
+			? auth_step_e::server_final
+			: auth_step_e::server_challenge;
+		auto data = auth_props[prop::authentication_data].value_or("");
+		return _auth.async_auth(
+			auth_step, std::move(data),
+			asio::prepend(std::move(*this), on_auth_data {}, auth_step)
+		);
+	}
+
+	void operator()(
+		on_auth_data, auth_step_e auth_step, error_code ec, std::string data
+	) {
+		if (ec)
+			return on_auth_fail(
+				"Re-authentication: authentication fail",
+				disconnect_rc_e::unspecified_error
+			);
+
+		if (auth_step == auth_step_e::server_final)
+			return;
+
+		auth_props props;
+		props[prop::authentication_method] = _auth.method();
+		props[prop::authentication_data] = std::move(data);
+		auto rc = auth_step == auth_step_e::client_initial
+			? reason_codes::reauthenticate
+			: reason_codes::continue_authentication;
+		auto packet = control_packet<allocator_type>::of(
+			no_pid, get_allocator(),
+			encoders::encode_auth,
+			rc.value(), props
+		);
+
+		const auto& wire_data = packet.wire_data();
+		_svc_ptr->async_send(
+			wire_data,
+			no_serial, send_flag::none,
+			asio::consign(asio::detached, std::move(packet))
+		);
+	}
+
+private:
+	void on_auth_fail(std::string message, disconnect_rc_e reason) {
+		auto props = disconnect_props{};
+		props[prop::reason_string] = std::move(message);
+		async_disconnect(
+			reason, props, false, _svc_ptr,
+			asio::detached
+		);
+	}
+
+};
+
+
+} // end namespace async_mqtt5::detail
+
+#endif // !ASYNC_MQTT5_RE_AUTH_OP_HPP

include/async_mqtt5/impl/read_message_op.hpp

@@ -12,6 +12,7 @@
 
 #include <async_mqtt5/impl/disconnect_op.hpp>
 #include <async_mqtt5/impl/publish_rec_op.hpp>
+#include <async_mqtt5/impl/re_auth_op.hpp>
 
 #include <async_mqtt5/impl/internal/codecs/message_decoders.hpp>
 
@@ -104,7 +105,15 @@ private:
 			}
 			break;
 			case auth: {
-				// TODO: dispatch auth
+				auto rv = decoders::decode_auth(
+					std::distance(first, last), first
+				);
+				if (!rv.has_value())
+					return on_malformed_packet(
+						"Malformed AUTH received: cannot decode"
+					);
+
+				re_auth_op { _svc_ptr }.perform(std::move(*rv));
 			}
 			break;
 		}

include/async_mqtt5/mqtt_client.hpp

@@ -11,6 +11,7 @@
 #include <async_mqtt5/impl/read_message_op.hpp>
 #include <async_mqtt5/impl/subscribe_op.hpp>
 #include <async_mqtt5/impl/unsubscribe_op.hpp>
+#include <async_mqtt5/impl/re_auth_op.hpp>
 
 namespace async_mqtt5 {
 
@@ -224,7 +225,7 @@ public:
 	/**
 	 * \brief Assign an authenticator that the Client will use for
 	 * \__ENHANCED_AUTH\__ on every connect to a Broker.
-	 * Re-authentication can be initiated by calling \ref async_authenticate.
+	 * Re-authentication can be initiated by calling \ref re_authenticate.
 	 *
 	 * \param authenticator Object that will be stored (move-constructed or by reference)
 	 * and used for authentication. It needs to satisfy \__is_authenticator\__ concept.
@@ -237,22 +238,13 @@ public:
 	}
 
 	/**
-	 * \brief Initiates re-authentication.
-	 * TODO
+	 * \brief Initiates [mqttlink 3901257 Re-authentication]
+	 * using the authenticator given in the \ref authenticator method.
+	 *
+	 * \note If \ref authenticator was not called, this method does nothing.
 	 */
-	template <typename CompletionToken>
-	decltype(auto) async_authenticate(CompletionToken&& token) {
-		using Signature = void(error_code);
-
-		auto initiate = [] (
-			auto handler
-		) {
-			// TODO re-authentication
-		};
-
-		return asio::async_initiate<CompletionToken, Signature>(
-			std::move(initiate), token
-		);
+	void re_authenticate() {
+		detail::re_auth_op { _svc_ptr }.perform();
 	}
 
 	/**
@@ -565,7 +557,7 @@ public:
 	 * Calling this function will attempt to receive an Application Message
 	 * from internal storage.
 	 *
-	 * \note The completion handler will be only invoked if an error occurred
+	 * \note The completion handler will only be invoked if an error occurred
 	 * or there is a pending Application Message.
 	 *
 	 * \param token Completion token that will be used to produce a

test/unit/include/test_common/test_broker.hpp

@@ -14,8 +14,6 @@
 
 #include <boost/asio/ip/tcp.hpp>
 
-#include <boost/test/included/unit_test.hpp>
-
 #include <async_mqtt5/impl/internal/codecs/message_decoders.hpp>
 
 #include "test_common/message_exchange.hpp"

test/unit/src/run_tests.cpp

@@ -1,4 +1,4 @@
-#include <boost/test/unit_test.hpp>
+#include <boost/test/included/unit_test.hpp>
 
 #include <test_common/protocol_logging.hpp>