Ported the config for SSL and websocket into new broker

2018-11-02 17:21:35 +01:00
parent 176b986c47
commit 5efbe8e8bf
9 changed files with 216 additions and 153 deletions
@@ -1,6 +1,5 @@
 package io.moquette.broker;

-import io.moquette.spi.ClientSession;
 import io.moquette.spi.impl.subscriptions.Topic;
 import io.moquette.spi.security.IAuthorizatorPolicy;
 import io.netty.handler.codec.mqtt.MqttQoS;
@@ -33,14 +33,22 @@ import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.codec.MessageToMessageDecoder;
 import io.netty.handler.codec.MessageToMessageEncoder;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpRequestDecoder;
+import io.netty.handler.codec.http.HttpResponseEncoder;
+import io.netty.handler.codec.http.HttpServerCodec;
 import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
 import io.netty.handler.codec.mqtt.MqttDecoder;
 import io.netty.handler.codec.mqtt.MqttEncoder;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslHandler;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.concurrent.Future;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

+import javax.net.ssl.SSLEngine;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
@@ -144,18 +152,18 @@ public class NewNettyAcceptor {
            this.errorsCather = Optional.empty();
        }
        initializePlainTCPTransport(mqttHandler, props);
-//        initializeWebSocketTransport(mqttHandler, props);
-//        String sslTcpPortProp = props.getProperty(BrokerConstants.SSL_PORT_PROPERTY_NAME);
-//        String wssPortProp = props.getProperty(BrokerConstants.WSS_PORT_PROPERTY_NAME);
-//        if (sslTcpPortProp != null || wssPortProp != null) {
-//            SslContext sslContext = sslCtxCreator.initSSLContext();
-//            if (sslContext == null) {
-//                LOG.error("Can't initialize SSLHandler layer! Exiting, check your configuration of jks");
-//                return;
-//            }
-//            initializeSSLTCPTransport(mqttHandler, props, sslContext);
-//            initializeWSSTransport(mqttHandler, props, sslContext);
-//        }
+        initializeWebSocketTransport(mqttHandler, props);
+        String sslTcpPortProp = props.getProperty(BrokerConstants.SSL_PORT_PROPERTY_NAME);
+        String wssPortProp = props.getProperty(BrokerConstants.WSS_PORT_PROPERTY_NAME);
+        if (sslTcpPortProp != null || wssPortProp != null) {
+            SslContext sslContext = sslCtxCreator.initSSLContext();
+            if (sslContext == null) {
+                LOG.error("Can't initialize SSLHandler layer! Exiting, check your configuration of jks");
+                return;
+            }
+            initializeSSLTCPTransport(mqttHandler, props, sslContext);
+            initializeWSSTransport(mqttHandler, props, sslContext);
+        }
    }

    private void initFactory(String host, int port, String protocol, final PipelineInitializer pipeliner) {
@@ -219,117 +227,117 @@ public class NewNettyAcceptor {
        });
    }

-//    private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) {
-//        LOG.debug("Configuring Websocket MQTT transport");
-//        String webSocketPortProp = props.getProperty(WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
-//        if (DISABLED_PORT_BIND.equals(webSocketPortProp)) {
-//            // Do nothing no WebSocket configured
-//            LOG.info("Property {} has been setted to {}. Websocket MQTT will be disabled",
-//                    BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
-//            return;
-//        }
-//        int port = Integer.parseInt(webSocketPortProp);
-//
-//        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
-//
-//        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
-//        initFactory(host, port, "Websocket MQTT", new PipelineInitializer() {
-//
-//            @Override
-//            void init(SocketChannel channel) {
-//                ChannelPipeline pipeline = channel.pipeline();
-//                pipeline.addLast(new HttpServerCodec());
-//                pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
-//                pipeline.addLast("webSocketHandler",
-//                        new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST));
-//                pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
-//                pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
-//                pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
-//                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
-//                pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
-//                pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
-//                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
-//                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
-//                pipeline.addLast("messageLogger", new MQTTMessageLogger());
-//                pipeline.addLast("handler", handler);
-//            }
-//        });
-//    }
-//
-//    private void initializeSSLTCPTransport(NettyMQTTHandler handler, IConfig props, SslContext sslContext) {
-//        LOG.debug("Configuring SSL MQTT transport");
-//        String sslPortProp = props.getProperty(SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
-//        if (DISABLED_PORT_BIND.equals(sslPortProp)) {
-//            // Do nothing no SSL configured
-//            LOG.info("Property {} has been set to {}. SSL MQTT will be disabled",
-//                    BrokerConstants.SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
-//            return;
-//        }
-//
-//        int sslPort = Integer.parseInt(sslPortProp);
-//        LOG.debug("Starting SSL on port {}", sslPort);
-//
-//        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
-//        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
-//        String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
-//        final boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
-//        initFactory(host, sslPort, "SSL MQTT", new PipelineInitializer() {
-//
-//            @Override
-//            void init(SocketChannel channel) throws Exception {
-//                ChannelPipeline pipeline = channel.pipeline();
-//                pipeline.addLast("ssl", createSslHandler(channel, sslContext, needsClientAuth));
-//                pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
-//                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
-//                // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
-//                pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
-//                pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
-//                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
-//                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
-//                pipeline.addLast("messageLogger", new MQTTMessageLogger());
-//                pipeline.addLast("handler", handler);
-//            }
-//        });
-//    }
-//
-//    private void initializeWSSTransport(NettyMQTTHandler handler, IConfig props, SslContext sslContext) {
-//        LOG.debug("Configuring secure websocket MQTT transport");
-//        String sslPortProp = props.getProperty(WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
-//        if (DISABLED_PORT_BIND.equals(sslPortProp)) {
-//            // Do nothing no SSL configured
-//            LOG.info("Property {} has been set to {}. Secure websocket MQTT will be disabled",
-//                    BrokerConstants.WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
-//            return;
-//        }
-//        int sslPort = Integer.parseInt(sslPortProp);
-//        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
-//        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
-//        String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
-//        final boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
-//        initFactory(host, sslPort, "Secure websocket", new PipelineInitializer() {
-//
-//            @Override
-//            void init(SocketChannel channel) throws Exception {
-//                ChannelPipeline pipeline = channel.pipeline();
-//                pipeline.addLast("ssl", createSslHandler(channel, sslContext, needsClientAuth));
-//                pipeline.addLast("httpEncoder", new HttpResponseEncoder());
-//                pipeline.addLast("httpDecoder", new HttpRequestDecoder());
-//                pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
-//                pipeline.addLast("webSocketHandler",
-//                        new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST));
-//                pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
-//                pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
-//                pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
-//                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
-//                pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
-//                pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
-//                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
-//                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
-//                pipeline.addLast("messageLogger", new MQTTMessageLogger());
-//                pipeline.addLast("handler", handler);
-//            }
-//        });
-//    }
+    private void initializeWebSocketTransport(final NewNettyMQTTHandler handler, IConfig props) {
+        LOG.debug("Configuring Websocket MQTT transport");
+        String webSocketPortProp = props.getProperty(WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+        if (DISABLED_PORT_BIND.equals(webSocketPortProp)) {
+            // Do nothing no WebSocket configured
+            LOG.info("Property {} has been setted to {}. Websocket MQTT will be disabled",
+                    BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+            return;
+        }
+        int port = Integer.parseInt(webSocketPortProp);
+
+        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+
+        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
+        initFactory(host, port, "Websocket MQTT", new PipelineInitializer() {
+
+            @Override
+            void init(SocketChannel channel) {
+                ChannelPipeline pipeline = channel.pipeline();
+                pipeline.addLast(new HttpServerCodec());
+                pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
+                pipeline.addLast("webSocketHandler",
+                        new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST));
+                pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
+                pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
+                pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
+                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
+                pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
+                pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
+                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
+                pipeline.addLast("messageLogger", new MQTTMessageLogger());
+                pipeline.addLast("handler", handler);
+            }
+        });
+    }
+
+    private void initializeSSLTCPTransport(NewNettyMQTTHandler handler, IConfig props, SslContext sslContext) {
+        LOG.debug("Configuring SSL MQTT transport");
+        String sslPortProp = props.getProperty(SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+        if (DISABLED_PORT_BIND.equals(sslPortProp)) {
+            // Do nothing no SSL configured
+            LOG.info("Property {} has been set to {}. SSL MQTT will be disabled",
+                    BrokerConstants.SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+            return;
+        }
+
+        int sslPort = Integer.parseInt(sslPortProp);
+        LOG.debug("Starting SSL on port {}", sslPort);
+
+        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
+        String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
+        final boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
+        initFactory(host, sslPort, "SSL MQTT", new PipelineInitializer() {
+
+            @Override
+            void init(SocketChannel channel) throws Exception {
+                ChannelPipeline pipeline = channel.pipeline();
+                pipeline.addLast("ssl", createSslHandler(channel, sslContext, needsClientAuth));
+                pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
+                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
+                // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
+                pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
+                pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
+                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
+                pipeline.addLast("messageLogger", new MQTTMessageLogger());
+                pipeline.addLast("handler", handler);
+            }
+        });
+    }
+
+    private void initializeWSSTransport(NewNettyMQTTHandler handler, IConfig props, SslContext sslContext) {
+        LOG.debug("Configuring secure websocket MQTT transport");
+        String sslPortProp = props.getProperty(WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+        if (DISABLED_PORT_BIND.equals(sslPortProp)) {
+            // Do nothing no SSL configured
+            LOG.info("Property {} has been set to {}. Secure websocket MQTT will be disabled",
+                    BrokerConstants.WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+            return;
+        }
+        int sslPort = Integer.parseInt(sslPortProp);
+        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
+        String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
+        final boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
+        initFactory(host, sslPort, "Secure websocket", new PipelineInitializer() {
+
+            @Override
+            void init(SocketChannel channel) throws Exception {
+                ChannelPipeline pipeline = channel.pipeline();
+                pipeline.addLast("ssl", createSslHandler(channel, sslContext, needsClientAuth));
+                pipeline.addLast("httpEncoder", new HttpResponseEncoder());
+                pipeline.addLast("httpDecoder", new HttpRequestDecoder());
+                pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
+                pipeline.addLast("webSocketHandler",
+                        new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST));
+                pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
+                pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
+                pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
+                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
+                pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
+                pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
+                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
+                pipeline.addLast("messageLogger", new MQTTMessageLogger());
+                pipeline.addLast("handler", handler);
+            }
+        });
+    }

    @SuppressWarnings("FutureReturnValueIgnored")
    public void close() {
@@ -369,15 +377,15 @@ public class NewNettyAcceptor {
                metrics.messagesWrote(), bytesMetrics.readBytes(), bytesMetrics.wroteBytes());
    }

-//    private ChannelHandler createSslHandler(SocketChannel channel, SslContext sslContext, boolean needsClientAuth) {
-//        SSLEngine sslEngine = sslContext.newEngine(
-//                channel.alloc(),
-//                channel.remoteAddress().getHostString(),
-//                channel.remoteAddress().getPort());
-//        sslEngine.setUseClientMode(false);
-//        if (needsClientAuth) {
-//            sslEngine.setNeedClientAuth(true);
-//        }
-//        return new SslHandler(sslEngine);
-//    }
+    private ChannelHandler createSslHandler(SocketChannel channel, SslContext sslContext, boolean needsClientAuth) {
+        SSLEngine sslEngine = sslContext.newEngine(
+                channel.alloc(),
+                channel.remoteAddress().getHostString(),
+                channel.remoteAddress().getPort());
+        sslEngine.setUseClientMode(false);
+        if (needsClientAuth) {
+            sslEngine.setNeedClientAuth(true);
+        }
+        return new SslHandler(sslEngine);
+    }
 }
@@ -3,12 +3,12 @@ package io.moquette.broker;
 import io.moquette.BrokerConstants;
 import io.moquette.interception.InterceptHandler;
 import io.moquette.persistence.MemoryStorageService;
+import io.moquette.server.DefaultMoquetteSslContextCreator;
 import io.moquette.server.config.*;
 import io.moquette.spi.ISessionsStore;
 import io.moquette.spi.impl.BrokerInterceptor;
 import io.moquette.spi.impl.SessionsRepository;
-import io.moquette.spi.impl.security.AcceptAllAuthenticator;
-import io.moquette.spi.impl.security.PermitAllAuthorizatorPolicy;
+import io.moquette.spi.impl.security.*;
 import io.moquette.spi.impl.subscriptions.CTrieSubscriptionDirectory;
 import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
 import io.moquette.spi.security.IAuthenticator;
@@ -21,6 +21,7 @@ import org.slf4j.LoggerFactory;
 import java.io.File;
 import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
+import java.text.ParseException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -136,24 +137,17 @@ public class Server {
            config.setProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, handlerProp);
        }
        final String persistencePath = config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME);
-        LOG.debug("Configuring Using persistent store file, path={}", persistencePath);
-//        m_processorBootstrapper = new ProtocolProcessorBootstrapper();
-//        final ProtocolProcessor processor = m_processorBootstrapper.init(config, handlers, authenticator, authorizator,
-//            this);
+        LOG.debug("Configuring Using persistent store file, path: {}", persistencePath);
        initInterceptors(config, handlers);
        LOG.debug("Initialized MQTT protocol processor");
-//        if (sslCtxCreator == null) {
-//            LOG.info("Using default SSL context creator");
-//            sslCtxCreator = new DefaultMoquetteSslContextCreator(config);
-//        }
-        // DBG
-        if (authenticator == null) {
-            authenticator = new AcceptAllAuthenticator();
-        }
-        if (authorizatorPolicy == null) {
-            authorizatorPolicy = new PermitAllAuthorizatorPolicy();
+        if (sslCtxCreator == null) {
+            LOG.info("Using default SSL context creator");
+            sslCtxCreator = new DefaultMoquetteSslContextCreator(config);
        }
+        authenticator = initializeAuthenticator(authenticator, config);
+        authorizatorPolicy = initializeAuthorizatorPolicy(authorizatorPolicy, config);

+        // TODO user real implementation DBG
        MemoryStorageService memStorage = new MemoryStorageService(null, null);
        ISessionsStore sessionStore = memStorage.sessionsStore();
        SessionsRepository sessionsRepository = new SessionsRepository(sessionStore, null);
@@ -177,6 +171,63 @@ public class Server {
        initialized = true;
    }

+    private IAuthorizatorPolicy initializeAuthorizatorPolicy(IAuthorizatorPolicy authorizatorPolicy, IConfig props) {
+//        if (authorizatorPolicy == null) {
+//            authorizatorPolicy = new PermitAllAuthorizatorPolicy();
+//        }
+//        return authorizatorPolicy;
+
+        LOG.debug("Configuring MQTT authorizator policy");
+        String authorizatorClassName = props.getProperty(BrokerConstants.AUTHORIZATOR_CLASS_NAME, "");
+        if (authorizatorPolicy == null && !authorizatorClassName.isEmpty()) {
+            authorizatorPolicy = loadClass(authorizatorClassName, IAuthorizatorPolicy.class, IConfig.class, props);
+        }
+
+        if (authorizatorPolicy == null) {
+            String aclFilePath = props.getProperty(BrokerConstants.ACL_FILE_PROPERTY_NAME, "");
+            if (aclFilePath != null && !aclFilePath.isEmpty()) {
+                authorizatorPolicy = new DenyAllAuthorizatorPolicy();
+                try {
+                    LOG.info("Parsing ACL file. Path = {}", aclFilePath);
+                    IResourceLoader resourceLoader = props.getResourceLoader();
+                    authorizatorPolicy = ACLFileParser.parse(resourceLoader.loadResource(aclFilePath));
+                } catch (ParseException pex) {
+                    LOG.error("Unable to parse ACL file. path=" + aclFilePath, pex);
+                }
+            } else {
+                authorizatorPolicy = new PermitAllAuthorizatorPolicy();
+            }
+            LOG.info("Authorizator policy {} instance will be used", authorizatorPolicy.getClass().getName());
+        }
+        return authorizatorPolicy;
+    }
+
+    private IAuthenticator initializeAuthenticator(IAuthenticator authenticator, IConfig props) {
+//        if (authenticator == null) {
+//            authenticator = new AcceptAllAuthenticator();
+//        }
+//        return authenticator;
+
+        LOG.debug("Configuring MQTT authenticator");
+        String authenticatorClassName = props.getProperty(BrokerConstants.AUTHENTICATOR_CLASS_NAME, "");
+
+        if (authenticator == null && !authenticatorClassName.isEmpty()) {
+            authenticator = loadClass(authenticatorClassName, IAuthenticator.class, IConfig.class, props);
+        }
+
+        IResourceLoader resourceLoader = props.getResourceLoader();
+        if (authenticator == null) {
+            String passwdPath = props.getProperty(BrokerConstants.PASSWORD_FILE_PROPERTY_NAME, "");
+            if (passwdPath.isEmpty()) {
+                authenticator = new AcceptAllAuthenticator();
+            } else {
+                authenticator = new ResourceAuthenticator(resourceLoader, passwdPath);
+            }
+            LOG.info("An {} authenticator instance will be used", authenticator.getClass().getName());
+        }
+        return authenticator;
+    }
+
    private void initInterceptors(IConfig props, List<? extends InterceptHandler> embeddedObservers) {
        LOG.info("Configuring message interceptors...");

@@ -49,13 +49,13 @@ import org.slf4j.LoggerFactory;
 * Moquette server implementation to load SSL certificate from local filesystem path configured in
 * config file.
 */
-class DefaultMoquetteSslContextCreator implements ISslContextCreator {
+public class DefaultMoquetteSslContextCreator implements ISslContextCreator {

    private static final Logger LOG = LoggerFactory.getLogger(DefaultMoquetteSslContextCreator.class);

    private final IConfig props;

-    DefaultMoquetteSslContextCreator(IConfig props) {
+    public DefaultMoquetteSslContextCreator(IConfig props) {
        this.props = Objects.requireNonNull(props);
    }

@@ -38,6 +38,7 @@ import java.util.Set;
 *     <li>Optionally, QoS 0 messages pending transmission to the Client.</li>
 * </ul>
 */
+@Deprecated
 public abstract class ClientSession {

    public final String clientID;
@@ -30,6 +30,7 @@ import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.ConcurrentMap;

+@Deprecated
 public class DurableSession extends ClientSession {

    class OutboundFlightZone {
@@ -16,6 +16,7 @@

 package io.moquette.spi;

+@Deprecated
 public class EnqueuedMessage {

    public final IMessagesStore.StoredMessage msg;
@@ -18,6 +18,7 @@ package io.moquette.spi;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;

+@Deprecated
 class InboundFlightZone {

    private final Map<Integer, IMessagesStore.StoredMessage> inboundFlightMessages = new ConcurrentHashMap<>();
@@ -32,6 +32,7 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;

+@Deprecated
 public class TransientSession extends ClientSession {

    private static final Logger LOG = LoggerFactory.getLogger(TransientSession.class);