Validate SUBACK/UNSUBACK reason code count (#432)

Author: sfod

source/v5/mqtt5_client.c

@@ -3512,15 +3512,39 @@ int aws_mqtt5_client_service_operational_state(struct aws_mqtt5_client_operation
     return AWS_OP_SUCCESS;
 }
 
+static bool s_aws_mqtt5_client_validate_ack_reason_code_count(
+    const struct aws_mqtt5_operation *operation,
+    enum aws_mqtt5_packet_type packet_type,
+    const void *packet_view) {
+
+    if (packet_type == AWS_MQTT5_PT_SUBACK) {
+        const struct aws_mqtt5_packet_subscribe_view *subscribe_view = operation->packet_view;
+        const struct aws_mqtt5_packet_suback_view *suback_view = packet_view;
+
+        return subscribe_view->subscription_count == suback_view->reason_code_count;
+    }
+
+    if (packet_type == AWS_MQTT5_PT_UNSUBACK) {
+        const struct aws_mqtt5_packet_unsubscribe_view *unsubscribe_view = operation->packet_view;
+        const struct aws_mqtt5_packet_unsuback_view *unsuback_view = packet_view;
+
+        return unsubscribe_view->topic_filter_count == unsuback_view->reason_code_count;
+    }
+
+    return true;
+}
+
 void aws_mqtt5_client_operational_state_handle_ack(
     struct aws_mqtt5_client_operational_state *client_operational_state,
     aws_mqtt5_packet_id_t packet_id,
     enum aws_mqtt5_packet_type packet_type,
     const void *packet_view,
     int error_code) {
 
+    struct aws_mqtt5_client *client = client_operational_state->client;
+
     if (packet_type == AWS_MQTT5_PT_PUBACK) {
-        aws_mqtt5_client_flow_control_state_on_puback(client_operational_state->client);
+        aws_mqtt5_client_flow_control_state_on_puback(client);
     }
 
     struct aws_hash_element *elem = NULL;
@@ -3530,23 +3554,45 @@ void aws_mqtt5_client_operational_state_handle_ack(
         AWS_LOGF_ERROR(
             AWS_LS_MQTT5_CLIENT,
             "id=%p: received an ACK for an unknown operation with id %d",
-            (void *)client_operational_state->client,
+            (void *)client,
             (int)packet_id);
         return;
     } else {
-        AWS_LOGF_TRACE(
-            AWS_LS_MQTT5_CLIENT,
-            "id=%p: Processing ACK with id %d",
-            (void *)client_operational_state->client,
-            (int)packet_id);
+        AWS_LOGF_TRACE(AWS_LS_MQTT5_CLIENT, "id=%p: Processing ACK with id %d", (void *)client, (int)packet_id);
     }
 
     struct aws_mqtt5_operation *operation = elem->value;
 
     aws_linked_list_remove(&operation->node);
     aws_hash_table_remove(&client_operational_state->unacked_operations_table, &packet_id, NULL, NULL);
 
-    s_complete_operation(client_operational_state->client, operation, error_code, packet_type, packet_view);
+    if (!s_aws_mqtt5_client_validate_ack_reason_code_count(operation, packet_type, packet_view)) {
+        AWS_LOGF_ERROR(
+            AWS_LS_MQTT5_CLIENT,
+            "id=%p: received a %s with a reason code count that does not match the acknowledged request; treating as a "
+            "protocol error and disconnecting",
+            (void *)client,
+            aws_mqtt5_packet_type_to_c_string(packet_type));
+
+        /*
+         * We must complete the operation explicitly rather than relying on the shutdown's operational-state reset:
+         * a subscribe/unsubscribe is retainable under the default offline queue policy, so the reset would requeue it
+         * for retry, potentially producing an endless disconnect/reconnect/resubscribe loop against a misbehaving
+         * broker.
+         */
+        s_complete_operation(client, operation, AWS_ERROR_MQTT5_DECODE_PROTOCOL_ERROR, AWS_MQTT5_PT_NONE, NULL);
+
+        if (s_should_client_disconnect_cleanly(client)) {
+            s_aws_mqtt5_client_shutdown_channel_clean(
+                client, AWS_ERROR_MQTT5_DECODE_PROTOCOL_ERROR, AWS_MQTT5_DRC_PROTOCOL_ERROR);
+        } else {
+            s_aws_mqtt5_client_shutdown_channel(client, AWS_ERROR_MQTT5_DECODE_PROTOCOL_ERROR);
+        }
+
+        return;
+    }
+
+    s_complete_operation(client, operation, error_code, packet_type, packet_view);
 }
 
 bool aws_mqtt5_client_are_negotiated_settings_valid(const struct aws_mqtt5_client *client) {

tests/v5/mqtt5_client_tests.c

@@ -5529,7 +5529,7 @@ static int s_aws_mqtt5_server_send_aliased_publish_sequence(
 
     struct aws_mqtt5_packet_suback_view suback_view = {
         .packet_id = subscribe_view->packet_id,
-        .reason_code_count = 1,
+        .reason_code_count = subscribe_view->subscription_count,
         .reason_codes = s_alias_reason_codes,
     };
 
@@ -5727,7 +5727,7 @@ static int s_aws_mqtt5_server_send_aliased_publish_failure(
 
     struct aws_mqtt5_packet_suback_view suback_view = {
         .packet_id = subscribe_view->packet_id,
-        .reason_code_count = 1,
+        .reason_code_count = subscribe_view->subscription_count,
         .reason_codes = s_alias_reason_codes,
     };
 

tests/v5/mqtt5_to_mqtt3_adapter_tests.c

@@ -3169,6 +3169,16 @@ static int s_mqtt5to3_adapter_subscribe_multi_oversized_suback_fn(struct aws_all
 
     s_wait_for_n_adapter_operation_events(&fixture, AWS_MQTT3_OET_SUBSCRIBE_COMPLETE, 1);
 
+    struct aws_mqtt3_operation_event expected_events[] = {
+        {
+            .type = AWS_MQTT3_OET_SUBSCRIBE_COMPLETE,
+            .error_code = AWS_ERROR_MQTT5_DECODE_PROTOCOL_ERROR,
+        },
+    };
+
+    ASSERT_SUCCESS(s_aws_mqtt5_to_mqtt3_adapter_test_fixture_verify_operation_sequence_contains(
+        &fixture, AWS_ARRAY_SIZE(expected_events), expected_events));
+
     aws_mqtt5_to_mqtt3_adapter_test_fixture_clean_up(&fixture);
     aws_mqtt_library_clean_up();
 
@@ -3885,7 +3895,7 @@ static int s_mqtt5_mock_server_handle_unsubscribe_unsuback_failure(
 
     struct aws_mqtt5_packet_unsuback_view unsuback_view = {
         .packet_id = unsubscribe_view->packet_id,
-        .reason_code_count = AWS_ARRAY_SIZE(mqtt5_unsuback_codes),
+        .reason_code_count = unsubscribe_view->topic_filter_count,
         .reason_codes = mqtt5_unsuback_codes,
     };