Fix mTls authorize bug (#1456)

Author: Technoboy-

mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/MQTTAuthenticationService.java

@@ -88,7 +88,7 @@ private Map<String, AuthenticationProvider> getAuthenticationProviders(List<Stri
     public AuthenticationResult authenticate(boolean fromProxy,
                                              SSLSession session, MqttConnectMessage connectMessage) {
         if (fromProxy) {
-            return new AuthenticationResult(true, null, null);
+            return AuthenticationResult.PASSED;
         }
         String authMethod = MqttMessageUtils.getAuthMethod(connectMessage);
         if (authMethod != null) {
@@ -161,6 +161,7 @@ public AuthenticationDataSource getAuthData(String authMethod, MqttConnectPayloa
     public static class AuthenticationResult {
 
         public static final AuthenticationResult FAILED = new AuthenticationResult(false, null, null);
+        public static final AuthenticationResult PASSED = new AuthenticationResult(true, null, null);
         private final boolean authenticated;
         private final String userRole;
         private final AuthenticationDataSource authData;

mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/MQTTProxyProtocolMethodProcessor.java

@@ -13,7 +13,7 @@
  */
 package io.streamnative.pulsar.handlers.mqtt.proxy;
 
-import static io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils.createMqttConnectMessage;
+import static io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils.createMqtt5ConnectMessage;
 import static io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils.createMqttPublishMessage;
 import static io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils.createMqttSubscribeMessage;
 import com.google.common.collect.Lists;
@@ -140,9 +140,12 @@ public void doProcessConnect(MqttAdapterMessage adapter, String userRole,
                 .processor(this)
                 .build();
         connection.sendConnAck();
-        MqttConnectMessage connectMessage = createMqttConnectMessage(msg, userRole);
-        msg = connectMessage;
-        connection.setConnectMessage(msg);
+
+        if (proxyConfig.isMqttAuthorizationEnabled()) {
+            MqttConnectMessage connectMessage = createMqtt5ConnectMessage(msg);
+            msg = connectMessage;
+            connection.setConnectMessage(msg);
+        }
 
         ConnectEvent connectEvent = ConnectEvent.builder()
                 .clientId(connection.getClientId())
@@ -163,8 +166,10 @@ public void processPublish(MqttAdapterMessage adapter) {
                 proxyConfig.getDefaultTenant(), proxyConfig.getDefaultNamespace(),
                 TopicDomain.getEnum(proxyConfig.getDefaultTopicDomain()));
         adapter.setClientId(connection.getClientId());
-        MqttPublishMessage mqttMessage = createMqttPublishMessage(msg, connection.getUserRole());
-        adapter.setMqttMessage(mqttMessage);
+        if (proxyConfig.isMqttAuthorizationEnabled()) {
+            MqttPublishMessage mqttMessage = createMqttPublishMessage(msg, connection.getUserRole());
+            adapter.setMqttMessage(mqttMessage);
+        }
         startPublish()
                 .thenCompose(__ ->  writeToBroker(pulsarTopicName, adapter))
                 .whenComplete((unused, ex) -> {
@@ -295,8 +300,10 @@ public void processSubscribe(final MqttAdapterMessage adapter) {
             log.debug("[Proxy Subscribe] [{}] msg: {}", clientId, msg);
         }
         registerTopicListener(adapter);
-        MqttSubscribeMessage mqttMessage = createMqttSubscribeMessage(msg, connection.getUserRole());
-        adapter.setMqttMessage(mqttMessage);
+        if (proxyConfig.isMqttAuthorizationEnabled()) {
+            MqttSubscribeMessage mqttMessage = createMqttSubscribeMessage(msg, connection.getUserRole());
+            adapter.setMqttMessage(mqttMessage);
+        }
         doSubscribe(adapter, false)
                 .exceptionally(ex -> {
                     Throwable realCause = FutureUtil.unwrapCompletionException(ex);

mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/support/MQTTBrokerProtocolMethodProcessor.java

@@ -15,6 +15,7 @@
 
 import static io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils.createWillMessage;
 import static io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils.getAuthenticationRole;
+import static io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils.getPacketId;
 import static io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils.pingResp;
 import static io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils.topicSubscriptions;
 import io.netty.channel.ChannelHandlerContext;
@@ -221,8 +222,7 @@ private CompletableFuture<Void> doUnauthorized(MqttAdapterMessage adapter) {
         log.error("[Publish] not authorized to topic={}, userRole={}, CId= {}",
                 msg.variableHeader().topicName(), connection.getUserRole(),
                 connection.getClientId());
-        int packetId = msg.variableHeader().packetId();
-        packetId = packetId == -1 ? 1 : packetId;
+        int packetId = getPacketId(msg.variableHeader().packetId());
         MqttPubAck.MqttPubErrorAckBuilder pubAckBuilder = MqttPubAck
                 .errorBuilder(connection.getProtocolVersion())
                 .packetId(packetId)

mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/utils/MqttMessageUtils.java

@@ -189,16 +189,12 @@ public static MqttPublishMessage createMqttWillMessage(WillMessage willMessage)
         return builder.build();
     }
 
-    public static MqttConnectMessage createMqttConnectMessage(MqttConnectMessage connectMessage,
-                                                              String authData) {
+    public static MqttConnectMessage createMqtt5ConnectMessage(MqttConnectMessage connectMessage) {
         final MqttConnectVariableHeader header = connectMessage.variableHeader();
-        MqttProperties properties = new MqttProperties();
-        properties.add(new MqttProperties.UserProperty(AUTHENTICATE_ROLE_KEY, authData));
         MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader(
                 MqttVersion.MQTT_5.protocolName(), MqttVersion.MQTT_5.protocolLevel(), header.hasUserName(),
                 header.hasPassword(), header.isWillRetain(), header.willQos(), header.isWillFlag(),
-                header.isCleanSession(), header.keepAliveTimeSeconds(), properties
-        );
+                header.isCleanSession(), header.keepAliveTimeSeconds(), connectMessage.variableHeader().properties());
         MqttConnectMessage newConnectMessage = new MqttConnectMessage(connectMessage.fixedHeader(), variableHeader,
                 connectMessage.payload());
         return newConnectMessage;
@@ -287,4 +283,11 @@ public static byte[] getAuthData(MqttConnectMessage connectMessage) {
                 .getProperty(MqttProperties.MqttPropertyType.AUTHENTICATION_DATA.value());
         return authDataProperty != null ? authDataProperty.value() : null;
     }
+
+    public static int getPacketId(int packetId) {
+        if (packetId < 1) {
+            return 1;
+        }
+        return packetId;
+    }
 }