[broker-67] Design of authentication mechanism

acl-groovy-dsl/src/main/groovy/javasabr/mqtt/acl/groovy/dsl/builder/AllOfBuilder.groovy

@@ -12,7 +12,7 @@ class AllOfBuilder extends ConditionBuilder {
     USER_NAME, CLIENT_ID, IP_ADDRESS
   }
 
-  private final Set<Identity> alreadySetIdentities = new HashSet<>()
+  private final Set<Identity> alreadySetIdentities = EnumSet.noneOf(Identity.class)
 
   @Override
   ConditionBuilder userName(ValueMatcher<String>... userNames) {

application/build.gradle

@@ -9,10 +9,13 @@ description = "Standard configuration of standalone version of MQTT Broker"
 
 dependencies {
   implementation projects.coreService
+  implementation projects.credentialsSourceFile
+  implementation projects.credentialsSourceDb
   implementation libs.rlib.logger.slf4j
   implementation libs.springboot.starter.core
   implementation libs.springboot.starter.log4j2
 
+  testImplementation libs.r2dbc.h2
   testImplementation projects.testSupport
   testImplementation testFixtures(projects.network)
 }

application/src/main/java/javasabr/mqtt/broker/application/config/AuthenticationSpringConfig.java

@@ -0,0 +1,96 @@
+package javasabr.mqtt.broker.application.config;
+
+import static io.r2dbc.postgresql.PostgresqlConnectionFactoryProvider.OPTIONS;
+import static io.r2dbc.spi.ConnectionFactoryOptions.DATABASE;
+import static io.r2dbc.spi.ConnectionFactoryOptions.DRIVER;
+import static io.r2dbc.spi.ConnectionFactoryOptions.HOST;
+import static io.r2dbc.spi.ConnectionFactoryOptions.PASSWORD;
+import static io.r2dbc.spi.ConnectionFactoryOptions.PORT;
+import static io.r2dbc.spi.ConnectionFactoryOptions.USER;
+
+import io.r2dbc.pool.ConnectionPool;
+import io.r2dbc.pool.ConnectionPoolConfiguration;
+import io.r2dbc.spi.ConnectionFactories;
+import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.ConnectionFactoryOptions;
+import java.util.List;
+import java.util.Map;
+import javasabr.mqtt.service.auth.AuthenticationService;
+import javasabr.mqtt.service.auth.DefaultAuthenticationService;
+import javasabr.mqtt.service.auth.PasswordBasedAuthenticationProvider;
+import javasabr.mqtt.service.auth.provider.AuthenticationProvider;
+import javasabr.mqtt.service.auth.source.CredentialSource;
+import javasabr.mqtt.service.auth.source.DatabaseProperties;
+import javasabr.mqtt.service.auth.source.FileCredentialsSource;
+import javasabr.mqtt.service.auth.source.R2dbcCredentialsSource;
+import javasabr.rlib.collections.dictionary.DictionaryFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.r2dbc.core.DatabaseClient;
+
+@Configuration(proxyBeanMethods = false)
+public class AuthenticationSpringConfig {
+
+  private static final String LOCK_TIMEOUT_OPTION = "lock_timeout";
+  private static final String STATEMENT_TIMEOUT_OPTION = "statement_timeout";
+
+  @Bean
+  ConnectionFactory connectionFactory(DatabaseProperties config) {
+    Map<String, String> timeoutOptions = Map.of(
+        LOCK_TIMEOUT_OPTION, config.lockTimeout(),
+        STATEMENT_TIMEOUT_OPTION, config.statementTimeout());
+    ConnectionFactoryOptions connectionFactoryOptions = ConnectionFactoryOptions
+        .builder()
+        .option(DRIVER, config.driver())
+        .option(HOST, config.host())
+        .option(PORT, config.port())
+        .option(USER, config.username())
+        .option(PASSWORD, config.password())
+        .option(DATABASE, config.name())
+        .option(OPTIONS, timeoutOptions)
+        .build();
+    ConnectionFactory connectionFactory = ConnectionFactories.get(connectionFactoryOptions);
+    ConnectionPoolConfiguration configuration = ConnectionPoolConfiguration
+        .builder(connectionFactory)
+        .maxIdleTime(config.maxIdleTime())
+        .maxSize(config.maxPoolSize())
+        .initialSize(config.initialPoolSize())
+        .build();
+    return new ConnectionPool(configuration);
+  }
+
+  @Bean
+  DatabaseClient databaseClient(ConnectionFactory connectionFactory) {
+    return DatabaseClient.create(connectionFactory);
+  }
+
+  @Bean
+  CredentialSource credentialSource(@Value("${credentials.source.file.name:credentials}") String fileName) {
+    return new FileCredentialsSource(fileName);
+  }
+
+  @Bean
+  CredentialSource dbCredentialSource(DatabaseClient connectionFactory, DatabaseProperties databaseProperties) {
+    return new R2dbcCredentialsSource(connectionFactory, databaseProperties.credentialsQuery());
+  }
+
+  @Bean
+  AuthenticationProvider passwordBasedAuthenticationProvider(CredentialSource credentialSource) {
+    return new PasswordBasedAuthenticationProvider(credentialSource);
+  }
+
+  @Bean
+  AuthenticationService authenticationService(
+      List<AuthenticationProvider> authenticationProviders,
+      @Value("${authentication.allow.anonymous:false}") boolean allowAnonymousAuth,
+      @Value("${authentication.provider.default:basic}") String defaultProviderName) {
+    var providers = DictionaryFactory.mutableRefToRefDictionary(String.class, AuthenticationProvider.class);
+    authenticationProviders.forEach(value -> providers.put(value.getAuthMethodName(), value));
+    AuthenticationProvider defaultProvider = providers.get(defaultProviderName);
+    if (defaultProvider == null) {
+      throw new IllegalArgumentException("[%s] authenticator provider not found".formatted(defaultProviderName));
+    }
+    return new DefaultAuthenticationService(providers.toReadOnly(), defaultProvider, allowAnonymousAuth);
+  }
+}

application/src/main/java/javasabr/mqtt/broker/application/config/CredentialsSourceDatabaseProperties.java

@@ -0,0 +1,20 @@
+package javasabr.mqtt.broker.application.config;
+
+import java.time.Duration;
+import javasabr.mqtt.service.auth.source.DatabaseProperties;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@ConfigurationProperties(prefix = "credentials.source.db")
+public record CredentialsSourceDatabaseProperties(
+    String username,
+    String password,
+    String driver,
+    String host,
+    int port,
+    String name,
+    String credentialsQuery,
+    Duration maxIdleTime,
+    int initialPoolSize,
+    int maxPoolSize,
+    String lockTimeout,
+    String statementTimeout) implements DatabaseProperties {}

application/src/main/java/javasabr/mqtt/broker/application/config/MqttBrokerSpringConfig.java

@@ -13,16 +13,14 @@
 import javasabr.mqtt.network.message.in.PublishMqttInMessage;
 import javasabr.mqtt.network.user.NetworkMqttUserFactory;
 import javasabr.mqtt.service.AuthorizationService;
-import javasabr.mqtt.service.AuthenticationService;
-import javasabr.mqtt.service.AuthorizationService;
 import javasabr.mqtt.service.ClientIdRegistry;
 import javasabr.mqtt.service.ConnectionService;
-import javasabr.mqtt.service.CredentialSource;
 import javasabr.mqtt.service.MessageOutFactoryService;
 import javasabr.mqtt.service.PublishDeliveringService;
 import javasabr.mqtt.service.PublishReceivingService;
 import javasabr.mqtt.service.SubscriptionService;
 import javasabr.mqtt.service.TopicService;
+import javasabr.mqtt.service.auth.AuthenticationService;
 import javasabr.mqtt.service.handler.client.ExternalNetworkMqttUserReleaseHandler;
 import javasabr.mqtt.service.impl.DefaultConnectionService;
 import javasabr.mqtt.service.impl.DefaultMessageOutFactoryService;
@@ -32,10 +30,8 @@
 import javasabr.mqtt.service.impl.DefaultTopicService;
 import javasabr.mqtt.service.impl.DisabledAuthorizationService;
 import javasabr.mqtt.service.impl.ExternalNetworkMqttUserFactory;
-import javasabr.mqtt.service.impl.FileCredentialsSource;
 import javasabr.mqtt.service.impl.InMemoryClientIdRegistry;
 import javasabr.mqtt.service.impl.InMemorySubscriptionService;
-import javasabr.mqtt.service.impl.SimpleAuthenticationService;
 import javasabr.mqtt.service.message.handler.MqttInMessageHandler;
 import javasabr.mqtt.service.message.handler.impl.ConnectInMqttInMessageHandler;
 import javasabr.mqtt.service.message.handler.impl.DisconnectMqttInMessageHandler;
@@ -72,13 +68,17 @@
 import lombok.CustomLog;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.context.event.ApplicationStartedEvent;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
 import org.springframework.core.env.Environment;
 
 @CustomLog
 @Configuration(proxyBeanMethods = false)
+@Import(AuthenticationSpringConfig.class)
+@EnableConfigurationProperties(CredentialsSourceDatabaseProperties.class)
 public class MqttBrokerSpringConfig {
 
   @Bean
@@ -87,7 +87,10 @@ ClientIdRegistry clientIdRegistry(Environment env) {
         env.getProperty(
             "client.id.available.chars",
             "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"),
-        env.getProperty("client.id.max.length", int.class, 36));
+        env.getProperty(
+            "client.id.max.length",
+            int.class,
+            36));
   }
 
   @Bean
@@ -96,19 +99,6 @@ MqttSessionService mqttSessionService(
     return new InMemoryMqttSessionService(cleanInterval);
   }
 
-  @Bean
-  CredentialSource credentialSource(
-      @Value("${credentials.source.file.name:credentials}") String fileName) {
-    return new FileCredentialsSource(fileName);
-  }
-
-  @Bean
-  AuthenticationService authenticationService(
-      CredentialSource credentialSource,
-      @Value("${authentication.allow.anonymous:false}") boolean allowAnonymousAuth) {
-    return new SimpleAuthenticationService(credentialSource, allowAnonymousAuth);
-  }
-
   @Bean
   AuthorizationService authorizationService() {
     return new DisabledAuthorizationService();
@@ -164,7 +154,7 @@ MqttInMessageHandler publishAckMqttInMessageHandler(MessageOutFactoryService mes
   MqttInMessageHandler publishCompleteMqttInMessageHandler(MessageOutFactoryService messageOutFactoryService) {
     return new PublishCompleteMqttInMessageHandler(messageOutFactoryService);
   }
-  
+
   @Bean
   PublishPayloadMqttInMessageFieldValidator publishPayloadMqttInMessageFieldValidator(
       MessageOutFactoryService messageOutFactoryService) {
@@ -176,7 +166,7 @@ PublishQosMqttInMessageFieldValidator publishQosMqttInMessageFieldValidator(
       MessageOutFactoryService messageOutFactoryService) {
     return new PublishQosMqttInMessageFieldValidator(messageOutFactoryService);
   }
-  
+
   @Bean
   PublishRetainMqttInMessageFieldValidator publishRetainMqttInMessageFieldValidator(
       MessageOutFactoryService messageOutFactoryService) {
@@ -188,13 +178,13 @@ PublishMessageExpiryIntervalMqttInMessageFieldValidator publishMessageExpiryInte
       MessageOutFactoryService messageOutFactoryService) {
     return new PublishMessageExpiryIntervalMqttInMessageFieldValidator(messageOutFactoryService);
   }
-  
+
   @Bean
   PublishResponseTopicMqttInMessageFieldValidator publishResponseTopicMqttInMessageFieldValidator(
       MessageOutFactoryService messageOutFactoryService) {
     return new PublishResponseTopicMqttInMessageFieldValidator(messageOutFactoryService);
   }
-  
+
   @Bean
   PublishTopicAliasMqttInMessageFieldValidator publishTopicAliasMqttInMessageFieldValidator(
       MessageOutFactoryService messageOutFactoryService) {
@@ -211,7 +201,8 @@ MqttInMessageHandler publishMqttInMessageHandler(
     return new PublishMqttInMessageHandler(
         publishReceivingService,
         messageOutFactoryService,
-        topicService, authorizationService,
+        topicService,
+        authorizationService,
         fieldValidators);
   }
 
@@ -235,7 +226,10 @@ MqttInMessageHandler subscribeMqttInMessageHandler(
       SubscriptionService subscriptionService,
       MessageOutFactoryService messageOutFactoryService,
       TopicService topicService) {
-    return new SubscribeMqttInMessageHandler(subscriptionService, messageOutFactoryService, topicService);
+    return new SubscribeMqttInMessageHandler(
+        subscriptionService,
+        messageOutFactoryService,
+        topicService);
   }
 
   @Bean
@@ -251,7 +245,9 @@ MqttInMessageHandler unsubscribeMqttInMessageHandler(
 
   @Bean
   ConnectionService externalMqttConnectionService(Collection<? extends MqttInMessageHandler> inMessageHandlers) {
-    return new DefaultConnectionService(ExternalNetworkMqttUser.class, inMessageHandlers);
+    return new DefaultConnectionService(
+        ExternalNetworkMqttUser.class,
+        inMessageHandlers);
   }
 
   @Bean
@@ -321,13 +317,19 @@ NetworkMqttUserReleaseHandler externalMqttClientReleaseHandler(
       ClientIdRegistry clientIdRegistry,
       MqttSessionService sessionService,
       SubscriptionService subscriptionService) {
-    return new ExternalNetworkMqttUserReleaseHandler(clientIdRegistry, sessionService, subscriptionService);
+    return new ExternalNetworkMqttUserReleaseHandler(
+        clientIdRegistry,
+        sessionService,
+        subscriptionService);
   }
 
   @Bean
   MqttServerConnectionConfig externalConnectionConfig(Environment env) {
     return new MqttServerConnectionConfig(
-        QoS.ofCode(env.getProperty("mqtt.connection.max.qos", int.class, 2)),
+        QoS.ofCode(env.getProperty(
+            "mqtt.connection.max.qos",
+            int.class,
+            2)),
         env.getProperty(
             "mqtt.external.connection.max.message.size",
             int.class,
@@ -371,15 +373,17 @@ MqttServerConnectionConfig externalConnectionConfig(Environment env) {
         env.getProperty(
             "mqtt.external.connection.retain.available",
             boolean.class,
-            false), // set false because currently it's not implemented and we should not allow for clients to use it
+            false),
+        // set false because currently it's not implemented and we should not allow for clients to use it
         env.getProperty(
             "mqtt.external.connection.wildcard.subscription.available",
             boolean.class,
             MqttProperties.WILDCARD_SUBSCRIPTION_AVAILABLE_DEFAULT),
         env.getProperty(
             "mqtt.external.connection.subscription.id.available",
             boolean.class,
-            false), // set false because currently it's not implemented and we should not allow for clients to use it
+            false),
+        // set false because currently it's not implemented and we should not allow for clients to use it
         env.getProperty(
             "mqtt.external.connection.shared.subscription.available",
             boolean.class,
@@ -393,14 +397,9 @@ ServerNetworkConfig externalNetworkConfig(
       @Value("${mqtt.external.network.write.buffer.size:512}") int writeBufferSize,
       @Value("${mqtt.external.network.thread.group.name:ExternalNetwork}") String threadGroupName,
       @Value("${mqtt.external.network.thread.count:1}") int threadGroupMaxSize) {
-    return ServerNetworkConfig.SimpleServerNetworkConfig
-        .builder()
-        .readBufferSize(readBufferSize)
-        .pendingBufferSize(pendingBufferSize)
-        .writeBufferSize(writeBufferSize)
-        .threadGroupName(threadGroupName)
-        .threadGroupMaxSize(threadGroupMaxSize)
-        .build();
+    return ServerNetworkConfig.SimpleServerNetworkConfig.builder().readBufferSize(readBufferSize)
+        .pendingBufferSize(pendingBufferSize).writeBufferSize(writeBufferSize).threadGroupName(threadGroupName)
+        .threadGroupMaxSize(threadGroupMaxSize).build();
   }
 
   @Bean
@@ -413,21 +412,28 @@ MqttConnectionFactory externalConnectionFactory(
       MqttServerConnectionConfig externalServerConnectionConfig,
       NetworkMqttUserFactory mqttUserFactory,
       @Value("${mqtt.external.connection.max.packets.by.read:100}") int maxPacketsByRead) {
-    return new DefaultMqttConnectionFactory(externalServerConnectionConfig, mqttUserFactory, maxPacketsByRead);
+    return new DefaultMqttConnectionFactory(
+        externalServerConnectionConfig,
+        mqttUserFactory,
+        maxPacketsByRead);
   }
 
   @Bean
   InetSocketAddress externalNetworkAddress(
       @Value("${mqtt.external.network.host:localhost}") String host,
       @Value("${mqtt.external.network.port:1883}") int port) {
-    return new InetSocketAddress(host, port);
+    return new InetSocketAddress(
+        host,
+        port);
   }
 
   @Bean
   ServerNetwork<MqttConnection> externalNetwork(
       ServerNetworkConfig externalNetworkConfig,
       MqttConnectionFactory externalConnectionFactory) {
-    return NetworkFactory.serverNetwork(externalNetworkConfig, externalConnectionFactory::newConnection);
+    return NetworkFactory.serverNetwork(
+        externalNetworkConfig,
+        externalConnectionFactory::newConnection);
   }
 
   @Bean
@@ -438,7 +444,9 @@ ApplicationListener<ApplicationStartedEvent> externalNetworkStarter(
     return _ -> {
       externalNetwork.start(externalNetworkAddress);
       externalNetwork.onAccept(connectionService::processAcceptedConnection);
-      log.info(externalNetworkAddress, "Started external MQTT network by address:[%s]"::formatted);
+      log.info(
+          externalNetworkAddress,
+          "Started external MQTT network by address:[%s]"::formatted);
     };
   }
 }