Support mTLS authentication for MoP

.github/workflows/pr_test.yml

@@ -100,7 +100,7 @@ jobs:
         run: ./scripts/retry.sh mvn -B -ntp test -Dtest=${{ matrix.test }} -DfailIfNoTests=false
 
       - name: Upload jacoco artifact
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v3
         with:
           name: ${{ matrix.test }}-jacoco-artifact
           path: '**/*.exec'

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

@@ -44,6 +44,7 @@
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import lombok.Getter;
+import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
 
@@ -69,7 +70,8 @@ public class Connection {
     @Getter
     private final TopicSubscriptionManager topicSubscriptionManager;
     @Getter
-    private final MqttConnectMessage connectMessage;
+    @Setter
+    private MqttConnectMessage connectMessage;
     @Getter
     private final ClientRestrictions clientRestrictions;
     @Getter

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

@@ -24,6 +24,8 @@ public final class Constants {
     public static final String AUTH_BASIC = "basic";
     public static final String AUTH_TOKEN = "token";
 
+    public static final String AUTH_MTLS = "mTls";
+
     public static final String ATTR_TOPIC_SUBS = "topicSubs";
 
     public static final String MQTT_PROPERTIES = "MQTT_PROPERTIES_%d_";

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

@@ -14,21 +14,25 @@
 package io.streamnative.pulsar.handlers.mqtt;
 
 import static io.streamnative.pulsar.handlers.mqtt.Constants.AUTH_BASIC;
+import static io.streamnative.pulsar.handlers.mqtt.Constants.AUTH_MTLS;
 import static io.streamnative.pulsar.handlers.mqtt.Constants.AUTH_TOKEN;
 import io.netty.handler.codec.mqtt.MqttConnectMessage;
 import io.netty.handler.codec.mqtt.MqttConnectPayload;
+import io.streamnative.pulsar.handlers.mqtt.identitypool.AuthenticationProviderMTls;
 import io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import javax.naming.AuthenticationException;
+import javax.net.ssl.SSLSession;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.authentication.AuthenticationDataCommand;
 import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
 import org.apache.pulsar.broker.authentication.AuthenticationProvider;
 import org.apache.pulsar.broker.authentication.AuthenticationService;
+import org.apache.pulsar.broker.service.BrokerService;
 
 /**
  * MQTT authentication service.
@@ -42,8 +46,11 @@ public class MQTTAuthenticationService {
     @Getter
     private final Map<String, AuthenticationProvider> authenticationProviders;
 
-    public MQTTAuthenticationService(AuthenticationService authenticationService, List<String> authenticationMethods) {
-        this.authenticationService = authenticationService;
+    private final BrokerService brokerService;
+
+    public MQTTAuthenticationService(BrokerService brokerService, List<String> authenticationMethods) {
+        this.brokerService = brokerService;
+        this.authenticationService = brokerService.getAuthenticationService();
         this.authenticationProviders = getAuthenticationProviders(authenticationMethods);
     }
 
@@ -53,6 +60,14 @@ private Map<String, AuthenticationProvider> getAuthenticationProviders(List<Stri
             final AuthenticationProvider authProvider = authenticationService.getAuthenticationProvider(method);
             if (authProvider != null) {
                 providers.put(method, authProvider);
+            } else if (AUTH_MTLS.equalsIgnoreCase(method)) {
+                AuthenticationProviderMTls providerMTls = new AuthenticationProviderMTls();
+                try {
+                    providerMTls.initialize(brokerService.pulsar().getLocalMetadataStore());
+                    providers.put(method, providerMTls);
+                } catch (Exception e) {
+                    log.error("Failed to initialize MQTT authentication method {} ", method, e);
+                }
             } else {
                 log.error("MQTT authentication method {} is not enabled in Pulsar configuration!", method);
             }
@@ -64,27 +79,32 @@ private Map<String, AuthenticationProvider> getAuthenticationProviders(List<Stri
         return providers;
     }
 
-    public AuthenticationResult authenticate(MqttConnectMessage connectMessage) {
+    public AuthenticationResult authenticate(boolean fromProxy,
+                                             SSLSession session, MqttConnectMessage connectMessage) {
         String authMethod = MqttMessageUtils.getAuthMethod(connectMessage);
         if (authMethod != null) {
             byte[] authData = MqttMessageUtils.getAuthData(connectMessage);
             if (authData == null) {
                 return AuthenticationResult.FAILED;
             }
+            if (fromProxy && AUTH_MTLS.equalsIgnoreCase(authMethod)) {
+                return new AuthenticationResult(true, new String(authData),
+                        new AuthenticationDataCommand(new String(authData), null, session));
+            }
             return authenticate(connectMessage.payload().clientIdentifier(), authMethod,
-                    new AuthenticationDataCommand(new String(authData)));
+                    new AuthenticationDataCommand(new String(authData), null, session));
         }
-        return authenticate(connectMessage.payload());
+        return authenticate(connectMessage.payload(), session);
     }
 
-    public AuthenticationResult authenticate(MqttConnectPayload payload) {
+    public AuthenticationResult authenticate(MqttConnectPayload payload, SSLSession session) {
         String userRole = null;
         boolean authenticated = false;
         AuthenticationDataSource authenticationDataSource = null;
         for (Map.Entry<String, AuthenticationProvider> entry : authenticationProviders.entrySet()) {
             String authMethod = entry.getKey();
             try {
-                AuthenticationDataSource authData = getAuthData(authMethod, payload);
+                AuthenticationDataSource authData = getAuthData(authMethod, payload, session);
                 userRole = entry.getValue().authenticate(authData);
                 authenticated = true;
                 authenticationDataSource = authData;
@@ -116,12 +136,14 @@ public AuthenticationResult authenticate(String clientIdentifier,
         return new AuthenticationResult(authenticated, userRole, command);
     }
 
-    public AuthenticationDataSource getAuthData(String authMethod, MqttConnectPayload payload) {
+    public AuthenticationDataSource getAuthData(String authMethod, MqttConnectPayload payload, SSLSession session) {
         switch (authMethod) {
             case AUTH_BASIC:
                 return new AuthenticationDataCommand(payload.userName() + ":" + payload.password());
             case AUTH_TOKEN:
                 return new AuthenticationDataCommand(payload.password());
+            case AUTH_MTLS:
+                return new AuthenticationDataCommand(null, null, session);
             default:
                 throw new IllegalArgumentException(
                         String.format("Unsupported authentication method : %s!", authMethod));

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

@@ -145,6 +145,13 @@ public class MQTTCommonConfiguration extends ServiceConfiguration {
     )
     private boolean mqttProxyTlsEnabled = false;
 
+    @FieldContext(
+            category = CATEGORY_MQTT_PROXY,
+            required = false,
+            doc = "Whether start mqtt protocol handler with proxy mtls"
+    )
+    private boolean mqttProxyMTlsAuthenticationEnabled = false;
+
     @FieldContext(
             category = CATEGORY_MQTT_PROXY,
             required = false,

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

@@ -96,7 +96,7 @@ public MQTTService(BrokerService brokerService, MQTTServerConfiguration serverCo
         this.metricsProvider = new MQTTMetricsProvider(metricsCollector);
         this.pulsarService.addPrometheusRawMetricsProvider(metricsProvider);
         this.authenticationService = serverConfiguration.isMqttAuthenticationEnabled()
-            ? new MQTTAuthenticationService(brokerService.getAuthenticationService(),
+            ? new MQTTAuthenticationService(brokerService,
                 serverConfiguration.getMqttAuthenticationMethods()) : null;
         this.connectionManager = new MQTTConnectionManager(pulsarService.getAdvertisedAddress());
         this.subscriptionManager = new MQTTSubscriptionManager();

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

@@ -51,7 +51,7 @@ public CompletableFuture<Void> writeAndFlush(final MqttAdapterMessage adapterMsg
         });
         future.exceptionally(ex -> {
             log.warn("[AdapterChannel][{}] Proxy write to broker {} failed."
-                    + " error message: {}", clientId, broker, ex.getMessage());
+                    + " adapterMsg message: {}", clientId, broker, adapterMsg, ex);
             return null;
         });
         return future;

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

@@ -0,0 +1,35 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamnative.pulsar.handlers.mqtt.exception;
+
+/**
+ * Internal server exception.
+ */
+public class MQTTAuthException extends Exception {
+
+    public MQTTAuthException() {
+    }
+
+    public MQTTAuthException(String message) {
+        super(message);
+    }
+
+    public MQTTAuthException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public MQTTAuthException(Throwable cause) {
+        super(cause);
+    }
+}

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

@@ -13,7 +13,6 @@
  */
 package io.streamnative.pulsar.handlers.mqtt.identitypool;
 
-
 import static io.streamnative.pulsar.handlers.mqtt.identitypool.ExpressionCompiler.DN;
 import static io.streamnative.pulsar.handlers.mqtt.identitypool.ExpressionCompiler.DN_KEYS;
 import static io.streamnative.pulsar.handlers.mqtt.identitypool.ExpressionCompiler.SAN;

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

@@ -13,6 +13,10 @@
  */
 package io.streamnative.pulsar.handlers.mqtt.proxy;
 
+import static io.streamnative.pulsar.handlers.mqtt.Constants.AUTH_MTLS;
+import static io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils.createMqttConnectMessage;
+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;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.mqtt.MqttConnectMessage;
@@ -64,17 +68,19 @@
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.util.Codec;
 import org.apache.pulsar.common.util.FutureUtil;
+
 /**
  * Proxy inbound handler is the bridge between proxy and MoP.
  */
 @Slf4j
 public class MQTTProxyProtocolMethodProcessor extends AbstractCommonProtocolMethodProcessor {
 
+    private final PulsarService pulsarService;
+
     @Getter
     private Connection connection;
     private final LookupHandler lookupHandler;
     private final MQTTProxyConfiguration proxyConfig;
-    private final PulsarService pulsarService;
     private final Map<String, CompletableFuture<AdapterChannel>> topicBrokers;
     private final Map<InetSocketAddress, AdapterChannel> adapterChannels;
     @Getter
@@ -86,6 +92,7 @@ public class MQTTProxyProtocolMethodProcessor extends AbstractCommonProtocolMeth
     private final MQTTConnectionManager connectionManager;
     private final SystemEventService eventService;
     private final MQTTProxyAdapter proxyAdapter;
+
     private final AtomicBoolean isDisconnected = new AtomicBoolean(false);
     private final AutoSubscribeHandler autoSubscribeHandler;
 
@@ -95,8 +102,9 @@ public class MQTTProxyProtocolMethodProcessor extends AbstractCommonProtocolMeth
 
     public MQTTProxyProtocolMethodProcessor(MQTTProxyService proxyService, ChannelHandlerContext ctx) {
         super(proxyService.getAuthenticationService(),
-                proxyService.getProxyConfig().isMqttAuthenticationEnabled(), ctx);
-        this.pulsarService = proxyService.getPulsarService();
+                proxyService.getProxyConfig().isMqttAuthenticationEnabled(),
+                ctx);
+        pulsarService = proxyService.getPulsarService();
         this.lookupHandler = proxyService.getLookupHandler();
         this.proxyConfig = proxyService.getProxyConfig();
         this.connectionManager = proxyService.getConnectionManager();
@@ -115,7 +123,7 @@ public MQTTProxyProtocolMethodProcessor(MQTTProxyService proxyService, ChannelHa
     @Override
     public void doProcessConnect(MqttAdapterMessage adapter, String userRole,
                                  AuthenticationDataSource authData, ClientRestrictions clientRestrictions) {
-        final MqttConnectMessage msg = (MqttConnectMessage) adapter.getMqttMessage();
+        MqttConnectMessage msg = (MqttConnectMessage) adapter.getMqttMessage();
         final ServerRestrictions serverRestrictions = ServerRestrictions.builder()
                 .receiveMaximum(proxyConfig.getReceiveMaximum())
                 .maximumPacketSize(proxyConfig.getMqttMessageMaxLength())
@@ -133,6 +141,12 @@ public void doProcessConnect(MqttAdapterMessage adapter, String userRole,
                 .processor(this)
                 .build();
         connection.sendConnAck();
+        if (proxyConfig.isMqttProxyMTlsAuthenticationEnabled()) {
+            MqttConnectMessage connectMessage = createMqttConnectMessage(msg, AUTH_MTLS, userRole);
+            msg = connectMessage;
+            connection.setConnectMessage(msg);
+        }
+
         ConnectEvent connectEvent = ConnectEvent.builder()
                 .clientId(connection.getClientId())
                 .address(pulsarService.getAdvertisedAddress())
@@ -152,6 +166,10 @@ public void processPublish(MqttAdapterMessage adapter) {
                 proxyConfig.getDefaultTenant(), proxyConfig.getDefaultNamespace(),
                 TopicDomain.getEnum(proxyConfig.getDefaultTopicDomain()));
         adapter.setClientId(connection.getClientId());
+        if (proxyConfig.isMqttProxyMTlsAuthenticationEnabled()) {
+            MqttPublishMessage mqttMessage = createMqttPublishMessage(msg, AUTH_MTLS, connection.getUserRole());
+            adapter.setMqttMessage(mqttMessage);
+        }
         startPublish()
                 .thenCompose(__ ->  writeToBroker(pulsarTopicName, adapter))
                 .whenComplete((unused, ex) -> {
@@ -282,6 +300,10 @@ public void processSubscribe(final MqttAdapterMessage adapter) {
             log.debug("[Proxy Subscribe] [{}] msg: {}", clientId, msg);
         }
         registerTopicListener(adapter);
+        if (proxyConfig.isMqttProxyMTlsAuthenticationEnabled()) {
+            MqttSubscribeMessage mqttMessage = createMqttSubscribeMessage(msg, AUTH_MTLS, connection.getUserRole());
+            adapter.setMqttMessage(mqttMessage);
+        }
         doSubscribe(adapter, false)
                 .exceptionally(ex -> {
                     Throwable realCause = FutureUtil.unwrapCompletionException(ex);
@@ -447,8 +469,10 @@ private CompletableFuture<AdapterChannel> connectToBroker(final String topic) {
                 key -> lookupHandler.findBroker(TopicName.get(topic)).thenApply(mqttBroker ->
                         adapterChannels.computeIfAbsent(mqttBroker, key1 -> {
                             AdapterChannel adapterChannel = proxyAdapter.getAdapterChannel(mqttBroker);
+                            final MqttConnectMessage connectMessage = connection.getConnectMessage();
+
                             adapterChannel.writeAndFlush(new MqttAdapterMessage(connection.getClientId(),
-                                connection.getConnectMessage()));
+                                    connectMessage));
                             return adapterChannel;
                         })
                 )

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

@@ -117,7 +117,7 @@ public void start() throws MQTTProxyException {
             throw new MQTTProxyException(e);
         }
 
-        if (proxyConfig.isMqttProxyTlsEnabled()) {
+        if (proxyConfig.isMqttProxyTlsEnabled() || proxyConfig.isMqttProxyMTlsAuthenticationEnabled()) {
             ServerBootstrap tlsBootstrap = serverBootstrap.clone();
             tlsBootstrap.childHandler(new MQTTProxyChannelInitializer(
                     this, proxyConfig, true, sslContextRefresher));
@@ -148,7 +148,6 @@ public void start() throws MQTTProxyException {
                 throw new MQTTProxyException(e);
             }
         }
-
         this.lookupHandler = new PulsarServiceLookupHandler(pulsarService, proxyConfig);
         this.eventService.start();
     }

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

@@ -21,10 +21,12 @@
 import io.netty.handler.codec.mqtt.MqttMessageBuilders;
 import io.netty.handler.codec.mqtt.MqttProperties;
 import io.netty.handler.codec.mqtt.MqttReasonCodeAndPropertiesVariableHeader;
+import io.netty.handler.ssl.SslHandler;
 import io.streamnative.pulsar.handlers.mqtt.Connection;
 import io.streamnative.pulsar.handlers.mqtt.MQTTAuthenticationService;
 import io.streamnative.pulsar.handlers.mqtt.ProtocolMethodProcessor;
 import io.streamnative.pulsar.handlers.mqtt.adapter.MqttAdapterMessage;
+import io.streamnative.pulsar.handlers.mqtt.exception.MQTTAuthException;
 import io.streamnative.pulsar.handlers.mqtt.exception.restrictions.InvalidReceiveMaximumException;
 import io.streamnative.pulsar.handlers.mqtt.messages.MqttPropertyUtils;
 import io.streamnative.pulsar.handlers.mqtt.messages.ack.MqttConnectAck;
@@ -33,6 +35,7 @@
 import io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils;
 import io.streamnative.pulsar.handlers.mqtt.utils.MqttUtils;
 import io.streamnative.pulsar.handlers.mqtt.utils.NettyUtils;
+import javax.net.ssl.SSLSession;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -123,8 +126,10 @@ public void processConnect(MqttAdapterMessage adapter) {
                         clientId, username);
             }
         } else {
-            MQTTAuthenticationService.AuthenticationResult authResult = authenticationService
-                    .authenticate(connectMessage);
+            MQTTAuthenticationService.AuthenticationResult authResult;
+            SslHandler sslHandler = ctx.pipeline().get(SslHandler.class);
+            SSLSession session = (sslHandler != null) ? sslHandler.engine().getSession() : null;
+            authResult = authenticationService.authenticate(adapter.fromProxy(), session, connectMessage);
             if (authResult.isFailed()) {
                 MqttMessage mqttMessage = MqttConnectAck.errorBuilder().authFail(protocolVersion);
                 log.error("[CONNECT] Invalid or incorrect authentication. CId={}, username={}", clientId, username);
@@ -157,6 +162,10 @@ public void processConnect(MqttAdapterMessage adapter) {
         }
     }
 
+    protected MQTTAuthenticationService.AuthenticationResult mtlsAuth(boolean fromProxy) throws MQTTAuthException {
+        return MQTTAuthenticationService.AuthenticationResult.FAILED;
+    }
+
     @Override
     public void processPubAck(MqttAdapterMessage msg) {
         if (log.isDebugEnabled()) {