Skip to content
Open
Show file tree
Hide file tree
Changes from 11 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
package javasabr.mqtt.service.acl


import javasabr.mqtt.model.acl.Operation
import javasabr.mqtt.model.acl.rule.Rule
import javasabr.mqtt.model.exception.AclConfigurationException
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
3 changes: 3 additions & 0 deletions application/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
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();
ConnectionPoolConfiguration configuration = ConnectionPoolConfiguration
.builder(ConnectionFactories.get(connectionFactoryOptions))
.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> credentialSource,
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

authenticationProviders or credentialSource?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed

@Value("${authentication.allow.anonymous:false}") boolean allowAnonymousAuth,
@Value("${authentication.provider.default:basic}") String defaultProviderName) {
var authenticationProviders = DictionaryFactory.mutableRefToRefDictionary(
String.class,
AuthenticationProvider.class);
credentialSource.forEach(value -> authenticationProviders.put(value.getAuthMethodName(), value));
AuthenticationProvider defaultProvider = authenticationProviders.get(defaultProviderName);
if (defaultProvider == null) {
throw new IllegalArgumentException("[%s] authenticator provider not found".formatted(defaultProviderName));
}
return new DefaultAuthenticationService(authenticationProviders.toReadOnly(), defaultProvider, allowAnonymousAuth);
}
}
Original file line number Diff line number Diff line change
@@ -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 {}
Original file line number Diff line number Diff line change
Expand Up @@ -13,15 +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.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;
Expand All @@ -31,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;
Expand Down Expand Up @@ -71,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
Expand All @@ -95,19 +96,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();
Expand Down Expand Up @@ -163,7 +151,7 @@ MqttInMessageHandler publishAckMqttInMessageHandler(MessageOutFactoryService mes
MqttInMessageHandler publishCompleteMqttInMessageHandler(MessageOutFactoryService messageOutFactoryService) {
return new PublishCompleteMqttInMessageHandler(messageOutFactoryService);
}

@Bean
PublishPayloadMqttInMessageFieldValidator publishPayloadMqttInMessageFieldValidator(
MessageOutFactoryService messageOutFactoryService) {
Expand All @@ -175,7 +163,7 @@ PublishQosMqttInMessageFieldValidator publishQosMqttInMessageFieldValidator(
MessageOutFactoryService messageOutFactoryService) {
return new PublishQosMqttInMessageFieldValidator(messageOutFactoryService);
}

@Bean
PublishRetainMqttInMessageFieldValidator publishRetainMqttInMessageFieldValidator(
MessageOutFactoryService messageOutFactoryService) {
Expand All @@ -187,13 +175,13 @@ PublishMessageExpiryIntervalMqttInMessageFieldValidator publishMessageExpiryInte
MessageOutFactoryService messageOutFactoryService) {
return new PublishMessageExpiryIntervalMqttInMessageFieldValidator(messageOutFactoryService);
}

@Bean
PublishResponseTopicMqttInMessageFieldValidator publishResponseTopicMqttInMessageFieldValidator(
MessageOutFactoryService messageOutFactoryService) {
return new PublishResponseTopicMqttInMessageFieldValidator(messageOutFactoryService);
}

@Bean
PublishTopicAliasMqttInMessageFieldValidator publishTopicAliasMqttInMessageFieldValidator(
MessageOutFactoryService messageOutFactoryService) {
Expand All @@ -210,7 +198,8 @@ MqttInMessageHandler publishMqttInMessageHandler(
return new PublishMqttInMessageHandler(
publishReceivingService,
messageOutFactoryService,
topicService, authorizationService,
topicService,
authorizationService,
fieldValidators);
}

Expand Down Expand Up @@ -242,10 +231,7 @@ MqttInMessageHandler unsubscribeMqttInMessageHandler(
SubscriptionService subscriptionService,
MessageOutFactoryService messageOutFactoryService,
TopicService topicService) {
return new UnsubscribeMqttInMessageHandler(
subscriptionService,
messageOutFactoryService,
topicService);
return new UnsubscribeMqttInMessageHandler(subscriptionService, messageOutFactoryService, topicService);
}

@Bean
Expand Down Expand Up @@ -285,32 +271,23 @@ MqttPublishInMessageHandler qos0MqttPublishInMessageHandler(
SubscriptionService subscriptionService,
PublishDeliveringService publishDeliveringService,
MessageOutFactoryService messageOutFactoryService) {
return new Qos0MqttPublishInMessageHandler(
subscriptionService,
publishDeliveringService,
messageOutFactoryService);
return new Qos0MqttPublishInMessageHandler(subscriptionService, publishDeliveringService, messageOutFactoryService);
}

@Bean
MqttPublishInMessageHandler qos1MqttPublishInMessageHandler(
SubscriptionService subscriptionService,
PublishDeliveringService publishDeliveringService,
MessageOutFactoryService messageOutFactoryService) {
return new Qos1MqttPublishInMessageHandler(
subscriptionService,
publishDeliveringService,
messageOutFactoryService);
return new Qos1MqttPublishInMessageHandler(subscriptionService, publishDeliveringService, messageOutFactoryService);
}

@Bean
MqttPublishInMessageHandler qos2MqttPublishInMessageHandler(
SubscriptionService subscriptionService,
PublishDeliveringService publishDeliveringService,
MessageOutFactoryService messageOutFactoryService) {
return new Qos2MqttPublishInMessageHandler(
subscriptionService,
publishDeliveringService,
messageOutFactoryService);
return new Qos2MqttPublishInMessageHandler(subscriptionService, publishDeliveringService, messageOutFactoryService);
}

@Bean
Expand All @@ -335,30 +312,15 @@ MqttServerConnectionConfig externalConnectionConfig(Environment env) {
"mqtt.external.connection.max.message.size",
int.class,
MqttProperties.MAXIMUM_MESSAGE_SIZE_DEFAULT),
env.getProperty(
"mqtt.external.connection.max.string.length",
int.class,
MqttProperties.MAXIMUM_STRING_LENGTH),
env.getProperty(
"mqtt.external.connection.max.binary.size",
int.class,
MqttProperties.MAXIMUM_BINARY_SIZE),
env.getProperty(
"mqtt.external.connection.max.topic.levels",
int.class,
MqttProperties.MAXIMUM_TOPIC_LEVELS),
env.getProperty(
"mqtt.external.connection.min.keep.alive",
int.class,
MqttProperties.SERVER_KEEP_ALIVE_DEFAULT),
env.getProperty("mqtt.external.connection.max.string.length", int.class, MqttProperties.MAXIMUM_STRING_LENGTH),
env.getProperty("mqtt.external.connection.max.binary.size", int.class, MqttProperties.MAXIMUM_BINARY_SIZE),
env.getProperty("mqtt.external.connection.max.topic.levels", int.class, MqttProperties.MAXIMUM_TOPIC_LEVELS),
env.getProperty("mqtt.external.connection.min.keep.alive", int.class, MqttProperties.SERVER_KEEP_ALIVE_DEFAULT),
env.getProperty(
"mqtt.external.connection.receive.maximum",
int.class,
MqttProperties.RECEIVE_MAXIMUM_PUBLISHES_DEFAULT),
env.getProperty(
"mqtt.external.connection.topic.alias.maximum",
int.class,
0),
env.getProperty("mqtt.external.connection.topic.alias.maximum", int.class, 0),
env.getProperty(
"mqtt.external.connection.default.session.expiration.time",
long.class,
Expand All @@ -371,18 +333,14 @@ MqttServerConnectionConfig externalConnectionConfig(Environment env) {
"mqtt.external.connection.sessions.enabled",
boolean.class,
MqttProperties.SESSIONS_ENABLED_DEFAULT),
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
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
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
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
env.getProperty(
"mqtt.external.connection.shared.subscription.available",
boolean.class,
Expand Down
Loading
Loading