fix topic authentication issue (#1734)

Author: Technoboy-

mqtt-broker/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/processor/MQTTBrokerProtocolMethodProcessor.java

@@ -91,6 +91,7 @@
 import org.apache.pulsar.broker.service.Subscription;
 import org.apache.pulsar.common.api.proto.CommandAck;
 import org.apache.pulsar.common.api.proto.CommandSubscribe;
+import org.apache.pulsar.common.naming.TopicDomain;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.util.Codec;
 import org.apache.pulsar.common.util.FutureUtil;
@@ -211,8 +212,8 @@ public void processPublish(MqttAdapterMessage adapter) {
                     userRole = authenticationRole.get();
                 }
             }
-            result = this.authorizationService.canProduceAsync(TopicName.get(msg.variableHeader().topicName()),
-                            userRole, authData)
+            result = this.authorizationService.canProduceAsync(
+                    getEncodedPulsarTopicName(msg.variableHeader().topicName()), userRole, authData)
                     .thenCompose(authorized -> authorized ? doPublish(adapter) : doUnauthorized(adapter));
         }
         result.thenAccept(__ -> msg.release())
@@ -225,6 +226,12 @@ public void processPublish(MqttAdapterMessage adapter) {
                 });
     }
 
+    private TopicName getEncodedPulsarTopicName(String mqttTopicName) {
+        return TopicName.get(PulsarTopicUtils.getEncodedPulsarTopicName(mqttTopicName,
+                configuration.getDefaultTenant(), configuration.getDefaultNamespace(),
+                TopicDomain.getEnum(configuration.getDefaultTopicDomain())));
+    }
+
     private CompletableFuture<Void> doUnauthorized(MqttAdapterMessage adapter) {
         final MqttPublishMessage msg = (MqttPublishMessage) adapter.getMqttMessage();
         log.error("[Publish] not authorized to topic={}, userRole={}, CId= {}",
@@ -385,8 +392,9 @@ public void processSubscribe(MqttAdapterMessage adapter) {
             AtomicBoolean authorizedFlag = new AtomicBoolean(true);
             for (MqttTopicSubscription topic: msg.payload().topicSubscriptions()) {
                 String finalUserRole = userRole;
-                authorizationFutures.add(this.authorizationService.canConsumeAsync(TopicName.get(topic.topicName()),
-                        userRole, authData, userRole).thenAccept((authorized) -> {
+                authorizationFutures.add(this.authorizationService.canConsumeAsync(
+                        getEncodedPulsarTopicName(topic.topicName()), userRole, authData, userRole)
+                        .thenAccept((authorized) -> {
                             if (!authorized) {
                                 authorizedFlag.set(false);
                                 log.warn("[Subscribe] no authorization to sub topic={}, userRole={}, CId= {}",

tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/broker/AdapterChannelTest.java

@@ -16,6 +16,7 @@
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertNotNull;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelId;
 import io.netty.handler.codec.mqtt.MqttConnectMessage;
@@ -84,7 +85,9 @@ public void testAdapterChannelAutoGetConnection() throws InterruptedException {
                         .connectMessage(fakeConnectMessage).build();
                 adapterChannel.writeAndFlush(connection, mqttAdapterMessage).join();
                 CompletableFuture<Channel> channelFutureAfterSend = brokerChannels.get(key);
+                assertNotNull(channelFutureAfterSend);
                 Channel channelAfterSend = channelFutureAfterSend.join();
+                assertNotNull(channelAfterSend);
                 assertNotEquals(channelAfterSend.id(), previousChannelId);
             }
         }

tests/src/test/java/io/streamnative/pulsar/handlers/mqtt/mqtt3/fusesource/base/AuthorizationTest.java

@@ -13,10 +13,13 @@
  */
 package io.streamnative.pulsar.handlers.mqtt.mqtt3.fusesource.base;
 
+import com.google.common.collect.Lists;
 import io.streamnative.pulsar.handlers.mqtt.base.AuthorizationConfig;
+import java.net.URLEncoder;
 import java.util.HashSet;
 import java.util.Set;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.admin.GrantTopicPermissionOptions;
 import org.apache.pulsar.common.policies.data.AuthAction;
 import org.awaitility.Awaitility;
 import org.fusesource.mqtt.client.BlockingConnection;
@@ -33,7 +36,7 @@
 public class AuthorizationTest extends AuthorizationConfig {
 
     @Test(timeOut = TIMEOUT)
-    public void testAuthorized() throws Exception {
+    public void testAuthorizedOnNamespace() throws Exception {
         Set<AuthAction> user1Actions = new HashSet<>();
         user1Actions.add(AuthAction.produce);
         admin.namespaces().grantPermissionOnNamespace("public/default", "user1", user1Actions);
@@ -65,6 +68,51 @@ public void testAuthorized() throws Exception {
         consumer.disconnect();
     }
 
+    @Test(timeOut = TIMEOUT)
+    public void testAuthorizedOnTopic() throws Exception {
+        String topicName = "persistent://public/default/testAuthorizedOnTopic/a";
+        String encodedTopicName = "persistent://public/default/" + URLEncoder.encode("testAuthorizedOnTopic/a");
+        admin.topics().createNonPartitionedTopic(encodedTopicName);
+        Set<AuthAction> user1Actions = new HashSet<>();
+        user1Actions.add(AuthAction.produce);
+        final GrantTopicPermissionOptions permission1 = GrantTopicPermissionOptions.builder()
+                .topic(encodedTopicName)
+                .role("user1")
+                .actions(user1Actions)
+                .build();
+        admin.namespaces().grantPermissionOnTopics(Lists.newArrayList(permission1));
+
+        Set<AuthAction> user2Actions = new HashSet<>();
+        user2Actions.add(AuthAction.consume);
+        final GrantTopicPermissionOptions permission2 = GrantTopicPermissionOptions.builder()
+                .topic(encodedTopicName)
+                .role("user2")
+                .actions(user2Actions)
+                .build();
+        admin.namespaces().grantPermissionOnTopics(Lists.newArrayList(permission2));
+
+        MQTT mqttConsumer = createMQTTClient();
+        mqttConsumer.setUserName("user2");
+        mqttConsumer.setPassword("pass2");
+        BlockingConnection consumer = mqttConsumer.blockingConnection();
+        consumer.connect();
+        Topic[] topics = {new Topic(topicName, QoS.AT_LEAST_ONCE)};
+        consumer.subscribe(topics);
+
+        MQTT mqttProducer = createMQTTClient();
+        mqttProducer.setUserName("user1");
+        mqttProducer.setPassword("pass1");
+        BlockingConnection producer = mqttProducer.blockingConnection();
+        producer.connect();
+        String message = "Hello MQTT";
+        producer.publish(topicName, message.getBytes(), QoS.AT_MOST_ONCE, false);
+
+        Message receive = consumer.receive();
+        Assert.assertEquals(new String(receive.getPayload()), message);
+        producer.disconnect();
+        consumer.disconnect();
+    }
+
     @Test
     public void testNotAuthorized() throws Exception {
         Set<AuthAction> user3Actions = new HashSet<>();