This closes #3102

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallback.java

@@ -107,7 +107,7 @@ public class AMQPConnectionCallback implements FailureListener, CloseListener {
       if (isPermittedMechanism(mechanism)) {
          switch (mechanism) {
             case PlainSASL.NAME:
-               result = new PlainSASL(server.getSecurityStore());
+               result = new PlainSASL(server.getSecurityStore(), manager.getSecurityDomain());
                break;
 
             case AnonymousServerSASL.NAME:

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java

@@ -206,7 +206,7 @@ public class AMQPSessionCallback implements SessionCallback {
                                                         false, // boolean autoCommitAcks,
                                                         false, // boolean preAcknowledge,
                                                         true, //boolean xa,
-                                                        (String) null, this, true, operationContext, manager.getPrefixes());
+                                                        (String) null, this, true, operationContext, manager.getPrefixes(), manager.getSecurityDomain());
    }
 
    @Override

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerReceiverContext.java

@@ -245,6 +245,11 @@ public class ProtonServerReceiverContext extends ProtonInitializable implements
                      public RemotingConnection getRemotingConnection() {
                         return connection.connectionCallback.getProtonConnectionDelegate();
                      }
+
+                     @Override
+                     public String getSecurityDomain() {
+                        return connection.getProtocolManager().getSecurityDomain();
+                     }
                   });
                } catch (ActiveMQSecurityException e) {
                   throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.securityErrorCreatingProducer(e.getMessage());

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/sasl/PlainSASL.java

@@ -21,16 +21,18 @@ import org.apache.activemq.artemis.core.security.SecurityStore;
 public class PlainSASL extends ServerSASLPlain {
 
    private final SecurityStore securityStore;
+   private final String securityDomain;
 
-   public PlainSASL(SecurityStore securityStore) {
+   public PlainSASL(SecurityStore securityStore, String securityDomain) {
       this.securityStore = securityStore;
+      this.securityDomain = securityDomain;
    }
 
    @Override
    protected boolean authenticate(String user, String password) {
       if (securityStore.isSecurityEnabled()) {
          try {
-            securityStore.authenticate(user, password, null);
+            securityStore.authenticate(user, password, null, securityDomain);
             return true;
          } catch (Exception e) {
             return false;

artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java

@@ -134,7 +134,8 @@ public class MQTTConnectionManager {
                                                          session.getSessionCallback(),
                                                          MQTTUtil.SESSION_AUTO_CREATE_QUEUE,
                                                          server.newOperationContext(),
-                                                         session.getProtocolManager().getPrefixes());
+                                                         session.getProtocolManager().getPrefixes(),
+                                                         session.getProtocolManager().getSecurityDomain());
       return (ServerSessionImpl) serverSession;
    }
 

artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConnection.java

@@ -235,6 +235,12 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se
       return this;
    }
 
+   // SecurityAuth implementation
+   @Override
+   public String getSecurityDomain() {
+      return protocolManager.getSecurityDomain();
+   }
+
    // SecurityAuth implementation
    @Override
    public String getPassword() {
@@ -759,7 +765,7 @@ public class OpenWireConnection extends AbstractRemotingConnection implements Se
    }
 
    private void createInternalSession(ConnectionInfo info) throws Exception {
-      internalSession = server.createSession(UUIDGenerator.getInstance().generateStringUUID(), context.getUserName(), info.getPassword(), -1, this, true, false, false, false, null, null, true, operationContext, protocolManager.getPrefixes());
+      internalSession = server.createSession(UUIDGenerator.getInstance().generateStringUUID(), context.getUserName(), info.getPassword(), -1, this, true, false, false, false, null, null, true, operationContext, protocolManager.getPrefixes(), protocolManager.getSecurityDomain());
    }
 
    //raise the refCount of context

artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManager.java

@@ -114,6 +114,8 @@ public class OpenWireProtocolManager implements ProtocolManager<Interceptor>, Cl
 
    private final ScheduledExecutorService scheduledPool;
 
+   private String securityDomain;
+
    //bean properties
    //http://activemq.apache.org/failover-transport-reference.html
    private boolean rebalanceClusterClients = false;
@@ -472,7 +474,7 @@ public class OpenWireProtocolManager implements ProtocolManager<Interceptor>, Cl
    }
 
    public void validateUser(String login, String passcode, OpenWireConnection connection) throws Exception {
-      server.getSecurityStore().authenticate(login, passcode, connection);
+      server.getSecurityStore().authenticate(login, passcode, connection, getSecurityDomain());
    }
 
    public void sendBrokerInfo(OpenWireConnection connection) throws Exception {
@@ -589,6 +591,16 @@ public class OpenWireProtocolManager implements ProtocolManager<Interceptor>, Cl
       return prefixes;
    }
 
+   @Override
+   public void setSecurityDomain(String securityDomain) {
+      this.securityDomain = securityDomain;
+   }
+
+   @Override
+   public String getSecurityDomain() {
+      return securityDomain;
+   }
+
    public List<DestinationInfo> getTemporaryDestinations() {
       List<DestinationInfo> total = new ArrayList<>();
       for (OpenWireConnection connection : connections) {

artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQSession.java

@@ -131,7 +131,7 @@ public class AMQSession implements SessionCallback {
       // now
 
       try {
-         coreSession = server.createSession(name, username, password, minLargeMessageSize, connection, true, false, false, false, null, this, true, connection.getOperationContext(), protocolManager.getPrefixes());
+         coreSession = server.createSession(name, username, password, minLargeMessageSize, connection, true, false, false, false, null, this, true, connection.getOperationContext(), protocolManager.getPrefixes(), protocolManager.getSecurityDomain());
 
          long sessionId = sessInfo.getSessionId().getValue();
          if (sessionId == -1) {

artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompProtocolManager.java

@@ -223,7 +223,7 @@ public class StompProtocolManager extends AbstractProtocolManager<StompFrame, St
       if (stompSession == null) {
          stompSession = new StompSession(connection, this, server.getStorageManager().newContext(server.getExecutorFactory().getExecutor()));
          String name = UUIDGenerator.getInstance().generateStringUUID();
-         ServerSession session = server.createSession(name, connection.getLogin(), connection.getPasscode(), ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, connection, !transacted, false, false, false, null, stompSession, true, server.newOperationContext(), getPrefixes());
+         ServerSession session = server.createSession(name, connection.getLogin(), connection.getPasscode(), ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, connection, !transacted, false, false, false, null, stompSession, true, server.newOperationContext(), getPrefixes(), getSecurityDomain());
          stompSession.setServerSession(session);
          sessions.put(id, stompSession);
       }

artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/AbstractControl.java

@@ -128,7 +128,7 @@ public abstract class AbstractControl extends StandardMBean {
                                                          Integer.MAX_VALUE, fakeConnection,
                                                          true, true, false,
                                                          false, address.toString(), fakeConnection.callback,
-                                                         false, new DummyOperationContext(), Collections.emptyMap());
+                                                         false, new DummyOperationContext(), Collections.emptyMap(), null);
       try {
          CoreMessage message = new CoreMessage(storageManager.generateID(), 50);
          if (headers != null) {

artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQPacketHandler.java

@@ -164,7 +164,7 @@ public class ActiveMQPacketHandler implements ChannelHandler {
          Map<SimpleString, RoutingType> routingTypeMap = protocolManager.getPrefixes();
 
          CoreSessionCallback sessionCallback = new CoreSessionCallback(request.getName(), protocolManager, channel, connection);
-         ServerSession session = server.createSession(request.getName(), activeMQPrincipal == null ? request.getUsername() : activeMQPrincipal.getUserName(), activeMQPrincipal == null ? request.getPassword() : activeMQPrincipal.getPassword(), request.getMinLargeMessageSize(), connection, request.isAutoCommitSends(), request.isAutoCommitAcks(), request.isPreAcknowledge(), request.isXA(), request.getDefaultAddress(), sessionCallback, true, sessionOperationContext, routingTypeMap);
+         ServerSession session = server.createSession(request.getName(), activeMQPrincipal == null ? request.getUsername() : activeMQPrincipal.getUserName(), activeMQPrincipal == null ? request.getPassword() : activeMQPrincipal.getPassword(), request.getMinLargeMessageSize(), connection, request.isAutoCommitSends(), request.isAutoCommitAcks(), request.isPreAcknowledge(), request.isXA(), request.getDefaultAddress(), sessionCallback, true, sessionOperationContext, routingTypeMap, protocolManager.getSecurityDomain());
          ServerProducer serverProducer = new ServerProducerImpl(session.getName(), "CORE", request.getDefaultAddress());
          session.addProducer(serverProducer);
          ServerSessionPacketHandler handler = new ServerSessionPacketHandler(server, protocolManager, session, server.getStorageManager(), channel);

artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/CoreProtocolManager.java

@@ -88,6 +88,8 @@ public class CoreProtocolManager implements ProtocolManager<Interceptor> {
 
    private final Map<SimpleString, RoutingType> prefixes = new HashMap<>();
 
+   private String securityDomain;
+
    public CoreProtocolManager(final CoreProtocolManagerFactory factory,
                               final ActiveMQServer server,
                               final List<Interceptor> incomingInterceptors,
@@ -221,6 +223,16 @@ public class CoreProtocolManager implements ProtocolManager<Interceptor> {
       return prefixes;
    }
 
+   @Override
+   public void setSecurityDomain(String securityDomain) {
+      this.securityDomain = securityDomain;
+   }
+
+   @Override
+   public String getSecurityDomain() {
+      return securityDomain;
+   }
+
    private boolean isArtemis(ActiveMQBuffer buffer) {
       return buffer.getByte(0) == 'A' &&
          buffer.getByte(1) == 'R' &&

artemis-server/src/main/java/org/apache/activemq/artemis/core/security/SecurityAuth.java

@@ -26,4 +26,6 @@ public interface SecurityAuth {
    String getPassword();
 
    RemotingConnection getRemotingConnection();
+
+   String getSecurityDomain();
 }

artemis-server/src/main/java/org/apache/activemq/artemis/core/security/SecurityStore.java

@@ -23,6 +23,8 @@ public interface SecurityStore {
 
    String authenticate(String user, String password, RemotingConnection remotingConnection) throws Exception;
 
+   String authenticate(String user, String password, RemotingConnection remotingConnection, String securityDomain) throws Exception;
+
    void check(SimpleString address, CheckType checkType, SecurityAuth session) throws Exception;
 
    void check(SimpleString address, SimpleString queue, CheckType checkType, SecurityAuth session) throws Exception;

artemis-server/src/main/java/org/apache/activemq/artemis/core/security/impl/SecurityStoreImpl.java

@@ -40,6 +40,7 @@ import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager;
 import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager2;
 import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager3;
+import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager4;
 import org.apache.activemq.artemis.utils.collections.ConcurrentHashSet;
 import org.apache.activemq.artemis.utils.collections.TypedProperties;
 import org.jboss.logging.Logger;
@@ -112,6 +113,14 @@ public class SecurityStoreImpl implements SecurityStore, HierarchicalRepositoryC
    public String authenticate(final String user,
                               final String password,
                               RemotingConnection connection) throws Exception {
+      return authenticate(user, password, connection, null);
+   }
+
+   @Override
+   public String authenticate(final String user,
+                              final String password,
+                              RemotingConnection connection,
+                              String securityDomain) throws Exception {
       if (securityEnabled) {
 
          if (managementClusterUser.equals(user)) {
@@ -133,7 +142,9 @@ public class SecurityStoreImpl implements SecurityStore, HierarchicalRepositoryC
          String validatedUser = null;
          boolean userIsValid = false;
 
-         if (securityManager instanceof ActiveMQSecurityManager3) {
+         if (securityManager instanceof ActiveMQSecurityManager4) {
+            validatedUser = ((ActiveMQSecurityManager4) securityManager).validateUser(user, password, connection, securityDomain);
+         } else if (securityManager instanceof ActiveMQSecurityManager3) {
             validatedUser = ((ActiveMQSecurityManager3) securityManager).validateUser(user, password, connection);
          } else if (securityManager instanceof ActiveMQSecurityManager2) {
             userIsValid = ((ActiveMQSecurityManager2) securityManager).validateUser(user, password, CertificateUtil.getCertsFromConnection(connection));
@@ -205,7 +216,10 @@ public class SecurityStoreImpl implements SecurityStore, HierarchicalRepositoryC
          }
 
          final boolean validated;
-         if (securityManager instanceof ActiveMQSecurityManager3) {
+         if (securityManager instanceof ActiveMQSecurityManager4) {
+            final ActiveMQSecurityManager4 securityManager4 = (ActiveMQSecurityManager4) securityManager;
+            validated = securityManager4.validateUserAndRole(user, session.getPassword(), roles, checkType, saddress, session.getRemotingConnection(), session.getSecurityDomain()) != null;
+         } else if (securityManager instanceof ActiveMQSecurityManager3) {
             final ActiveMQSecurityManager3 securityManager3 = (ActiveMQSecurityManager3) securityManager;
             validated = securityManager3.validateUserAndRole(user, session.getPassword(), roles, checkType, saddress, session.getRemotingConnection()) != null;
          } else if (securityManager instanceof ActiveMQSecurityManager2) {

artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java

@@ -319,7 +319,8 @@ public interface ActiveMQServer extends ServiceComponent {
                                SessionCallback callback,
                                boolean autoCreateQueues,
                                OperationContext context,
-                               Map<SimpleString, RoutingType> prefixes) throws Exception;
+                               Map<SimpleString, RoutingType> prefixes,
+                               String securityDomain) throws Exception;
 
    SecurityStore getSecurityStore();
 

artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java

@@ -1519,7 +1519,8 @@ public class ActiveMQServerImpl implements ActiveMQServer {
                                       final SessionCallback callback,
                                       final boolean autoCreateQueues,
                                       final OperationContext context,
-                                      final Map<SimpleString, RoutingType> prefixes) throws Exception {
+                                      final Map<SimpleString, RoutingType> prefixes,
+                                      final String securityDomain) throws Exception {
 
       if (AuditLogger.isEnabled()) {
          AuditLogger.createCoreSession(this, name, username, "****", minLargeMessageSize, connection, autoCommitSends,
@@ -1528,7 +1529,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
       String validatedUser = "";
 
       if (securityStore != null) {
-         validatedUser = securityStore.authenticate(username, password, connection);
+         validatedUser = securityStore.authenticate(username, password, connection, securityDomain);
       }
 
       checkSessionLimit(validatedUser);
@@ -1537,7 +1538,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
          callBrokerSessionPlugins(plugin -> plugin.beforeCreateSession(name, username, minLargeMessageSize, connection,
                  autoCommitSends, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, autoCreateQueues, context, prefixes));
       }
-      final ServerSessionImpl session = internalCreateSession(name, username, password, validatedUser, minLargeMessageSize, connection, autoCommitSends, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, context, autoCreateQueues, prefixes);
+      final ServerSessionImpl session = internalCreateSession(name, username, password, validatedUser, minLargeMessageSize, connection, autoCommitSends, autoCommitAcks, preAcknowledge, xa, defaultAddress, callback, context, autoCreateQueues, prefixes, securityDomain);
 
       sessions.put(name, session);
 
@@ -1614,8 +1615,9 @@ public class ActiveMQServerImpl implements ActiveMQServer {
                                                      SessionCallback callback,
                                                      OperationContext context,
                                                      boolean autoCreateJMSQueues,
-                                                     Map<SimpleString, RoutingType> prefixes) throws Exception {
+                                                     Map<SimpleString, RoutingType> prefixes,
-      return new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, configuration.isPersistDeliveryCountBeforeDelivery(), xa, connection, storageManager, postOffice, resourceManager, securityStore, managementService, this, configuration.getManagementAddress(), defaultAddress == null ? null : new SimpleString(defaultAddress), callback, context, pagingManager, prefixes);
+                                                     String securityDomain) throws Exception {
+      return new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, configuration.isPersistDeliveryCountBeforeDelivery(), xa, connection, storageManager, postOffice, resourceManager, securityStore, managementService, this, configuration.getManagementAddress(), defaultAddress == null ? null : new SimpleString(defaultAddress), callback, context, pagingManager, prefixes, securityDomain);
    }
 
    @Override

artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java

@@ -118,6 +118,8 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
 
    private boolean securityEnabled = true;
 
+   private final String securityDomain;
+
    protected final String username;
 
    protected final String password;
@@ -225,7 +227,8 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
                             final SessionCallback callback,
                             final OperationContext context,
                             final PagingManager pagingManager,
-                            final Map<SimpleString, RoutingType> prefixes) throws Exception {
+                            final Map<SimpleString, RoutingType> prefixes,
+                            final String securityDomain) throws Exception {
       this.username = username;
 
       this.password = password;
@@ -284,6 +287,8 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
       }
       //When the ServerSessionImpl initialization is complete, need to create and send a SESSION_CREATED notification.
       sendSessionNotification(CoreNotificationType.SESSION_CREATED);
+
+      this.securityDomain = securityDomain;
    }
 
    // ServerSession implementation ---------------------------------------------------------------------------
@@ -1064,6 +1069,11 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
       return remotingConnection;
    }
 
+   @Override
+   public String getSecurityDomain() {
+      return securityDomain;
+   }
+
    public static class TempQueueCleanerUpper implements CloseListener, FailureListener {
 
       private final SimpleString bindingName;

artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractProtocolManager.java

@@ -32,6 +32,8 @@ public abstract class AbstractProtocolManager<P, I extends BaseInterceptor<P>, C
 
    private final Map<SimpleString, RoutingType> prefixes = new HashMap<>();
 
+   private String securityDomain;
+
    protected String invokeInterceptors(final List<I> interceptors, final P message, final C connection) {
       if (interceptors != null && !interceptors.isEmpty()) {
          for (I interceptor : interceptors) {
@@ -66,4 +68,14 @@ public abstract class AbstractProtocolManager<P, I extends BaseInterceptor<P>, C
    public Map<SimpleString, RoutingType> getPrefixes() {
       return prefixes;
    }
+
+   @Override
+   public String getSecurityDomain() {
+      return securityDomain;
+   }
+
+   @Override
+   public void setSecurityDomain(String securityDomain) {
+      this.securityDomain = securityDomain;
+   }
 }

artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/ProtocolManager.java

@@ -74,4 +74,8 @@ public interface ProtocolManager<P extends BaseInterceptor> {
    void setMulticastPrefix(String multicastPrefix);
 
    Map<SimpleString, RoutingType> getPrefixes();
+
+   void setSecurityDomain(String securityDomain);
+
+   String getSecurityDomain();
 }

artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/ActiveMQJAASSecurityManager.java

@@ -44,7 +44,7 @@ import static org.apache.activemq.artemis.core.remoting.CertificateUtil.getCerts
  * The {@link Subject} returned by the login context is expecting to have a set of {@link RolePrincipal} for each
  * role of the user.
  */
-public class ActiveMQJAASSecurityManager implements ActiveMQSecurityManager3 {
+public class ActiveMQJAASSecurityManager implements ActiveMQSecurityManager4 {
 
    private static final Logger logger = Logger.getLogger(ActiveMQJAASSecurityManager.class);
 
@@ -90,13 +90,13 @@ public class ActiveMQJAASSecurityManager implements ActiveMQSecurityManager3 {
 
    @Override
    public boolean validateUser(String user, String password) {
-      throw new UnsupportedOperationException("Invoke validateUser(String, String, X509Certificate[]) instead");
+      throw new UnsupportedOperationException("Invoke validateUser(String, String, RemotingConnection, String) instead");
    }
 
    @Override
-   public String validateUser(final String user, final String password, RemotingConnection remotingConnection) {
+   public String validateUser(final String user, final String password, RemotingConnection remotingConnection, final String securityDomain) {
       try {
-         return getUserFromSubject(getAuthenticatedSubject(user, password, remotingConnection));
+         return getUserFromSubject(getAuthenticatedSubject(user, password, remotingConnection, securityDomain));
       } catch (LoginException e) {
          if (logger.isDebugEnabled()) {
             logger.debug("Couldn't validate user", e);
@@ -118,7 +118,7 @@ public class ActiveMQJAASSecurityManager implements ActiveMQSecurityManager3 {
 
    @Override
    public boolean validateUserAndRole(String user, String password, Set<Role> roles, CheckType checkType) {
-      throw new UnsupportedOperationException("Invoke validateUserAndRole(String, String, Set<Role>, CheckType, String, RemotingConnection) instead");
+      throw new UnsupportedOperationException("Invoke validateUserAndRole(String, String, Set<Role>, CheckType, String, RemotingConnection, String) instead");
    }
 
    @Override
@@ -127,10 +127,11 @@ public class ActiveMQJAASSecurityManager implements ActiveMQSecurityManager3 {
                                      final Set<Role> roles,
                                      final CheckType checkType,
                                      final String address,
-                                     final RemotingConnection remotingConnection) {
+                                     final RemotingConnection remotingConnection,
+                                     final String securityDomain) {
       Subject localSubject;
       try {
-         localSubject = getAuthenticatedSubject(user, password, remotingConnection);
+         localSubject = getAuthenticatedSubject(user, password, remotingConnection, securityDomain);
       } catch (LoginException e) {
          if (logger.isDebugEnabled()) {
             logger.debug("Couldn't validate user", e);
@@ -176,7 +177,8 @@ public class ActiveMQJAASSecurityManager implements ActiveMQSecurityManager3 {
 
    private Subject getAuthenticatedSubject(final String user,
                                            final String password,
-                                           final RemotingConnection remotingConnection) throws LoginException {
+                                           final RemotingConnection remotingConnection,
+                                           final String securityDomain) throws LoginException {
       LoginContext lc;
       ClassLoader currentLoader = Thread.currentThread().getContextClassLoader();
       ClassLoader thisLoader = this.getClass().getClassLoader();
@@ -184,7 +186,9 @@ public class ActiveMQJAASSecurityManager implements ActiveMQSecurityManager3 {
          if (thisLoader != currentLoader) {
             Thread.currentThread().setContextClassLoader(thisLoader);
          }
-         if (certificateConfigurationName != null && certificateConfigurationName.length() > 0 && getCertsFromConnection(remotingConnection) != null) {
+         if (securityDomain != null) {
+            lc = new LoginContext(securityDomain, null, new JaasCallbackHandler(user, password, remotingConnection), null);
+         } else if (certificateConfigurationName != null && certificateConfigurationName.length() > 0 && getCertsFromConnection(remotingConnection) != null) {
             lc = new LoginContext(certificateConfigurationName, null, new JaasCallbackHandler(user, password, remotingConnection), certificateConfiguration);
          } else {
             lc = new LoginContext(configurationName, null, new JaasCallbackHandler(user, password, remotingConnection), configuration);
@@ -307,5 +311,4 @@ public class ActiveMQJAASSecurityManager implements ActiveMQSecurityManager3 {
 
       return instance;
    }
-
 }

artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/ActiveMQSecurityManager4.java

@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.activemq.artemis.spi.core.security;
+
+import java.util.Set;
+
+import org.apache.activemq.artemis.core.security.CheckType;
+import org.apache.activemq.artemis.core.security.Role;
+import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
+
+/**
+ * Used to validate whether a user is authorized to connect to the
+ * server and perform certain functions on certain addresses
+ *
+ * This is an evolution of {@link ActiveMQSecurityManager3}
+ * that adds the ability to specify the JAAS domain per call.
+ */
+public interface ActiveMQSecurityManager4 extends ActiveMQSecurityManager {
+
+   /**
+    * is this a valid user.
+    *
+    * This method is called instead of
+    * {@link ActiveMQSecurityManager#validateUser(String, String)}.
+    *
+    * @param user     the user
+    * @param password the users password
+    * @param remotingConnection
+    * @param securityDomain the name of the JAAS security domain to use (can be null)
+    * @return the name of the validated user or null if the user isn't validated
+    */
+   String validateUser(String user, String password, RemotingConnection remotingConnection, String securityDomain);
+
+   /**
+    * Determine whether the given user is valid and whether they have
+    * the correct role for the given destination address.
+    *
+    * This method is called instead of
+    * {@link ActiveMQSecurityManager#validateUserAndRole(String, String, Set, CheckType)}.
+    *
+    * @param user       the user
+    * @param password   the user's password
+    * @param roles      the user's roles
+    * @param checkType  which permission to validate
+    * @param address    the address for which to perform authorization
+    * @param remotingConnection the user's connection
+    * @param securityDomain the name of the JAAS security domain to use (can be null)
+    * @return the name of the validated user or null if the user isn't validated
+    */
+   String validateUserAndRole(String user,
+                              String password,
+                              Set<Role> roles,
+                              CheckType checkType,
+                              String address,
+                              RemotingConnection remotingConnection,
+                              String securityDomain);
+}

artemis-server/src/test/java/org/apache/activemq/artemis/core/security/jaas/JAASSecurityManagerTest.java

@@ -80,7 +80,7 @@ public class JAASSecurityManagerTest {
          }
          ActiveMQJAASSecurityManager securityManager = new ActiveMQJAASSecurityManager("PropertiesLogin");
 
-         String result = securityManager.validateUser("first", "secret", null);
+         String result = securityManager.validateUser("first", "secret", null, null);
 
          assertNotNull(result);
          assertEquals("first", result);
@@ -88,7 +88,7 @@ public class JAASSecurityManagerTest {
          Role role = new Role("programmers", true, true, true, true, true, true, true, true, true, true);
          Set<Role> roles = new HashSet<>();
          roles.add(role);
-         result = securityManager.validateUserAndRole("first", "secret", roles, CheckType.SEND, "someaddress", null);
+         result = securityManager.validateUserAndRole("first", "secret", roles, CheckType.SEND, "someaddress", null, null);
 
          assertNotNull(result);
          assertEquals("first", result);

docs/user-manual/en/security.md

@@ -1253,3 +1253,16 @@ configure it in `bootstrap.xml` using the `security-manager` element, e.g.:
 ```
 
 The `security-manager` example demonstrates how to do this is more detail.
+
+## Per-Acceptor Security Domains
+
+It's possible to override the broker's JAAS security domain by specifying a
+security domain on an individual `acceptor`. Simply use the `securityDomain`
+parameter and indicate which domain from your `login.config` to use, e.g.:
+
+```xml
+<acceptor name="myAcceptor">tcp://127.0.0.1:61616?securityDomain=mySecurityDomain</acceptor>
+```
+
+Any client connecting to this acceptor will be have security enforced using
+`mySecurityDomain`.

examples/features/standard/security-manager/src/main/java/org/apache/activemq/artemis/jms/example/JAASSecurityManagerWrapper.java

@@ -25,15 +25,15 @@ import org.apache.activemq.artemis.core.security.Role;
 import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
 import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager;
-import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager3;
+import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager4;
 
-public class JAASSecurityManagerWrapper implements ActiveMQSecurityManager3 {
+public class JAASSecurityManagerWrapper implements ActiveMQSecurityManager4 {
    ActiveMQJAASSecurityManager activeMQJAASSecurityManager;
 
    @Override
-   public String validateUser(String user, String password, RemotingConnection remotingConnection) {
+   public String validateUser(String user, String password, RemotingConnection remotingConnection, String securityDomain) {
       System.out.println("validateUser(" + user + ", " + password + ", " + remotingConnection.getRemoteAddress() + ")");
-      return activeMQJAASSecurityManager.validateUser(user, password, remotingConnection);
+      return activeMQJAASSecurityManager.validateUser(user, password, remotingConnection, securityDomain);
    }
 
 
@@ -43,9 +43,10 @@ public class JAASSecurityManagerWrapper implements ActiveMQSecurityManager3 {
                               Set<Role> roles,
                               CheckType checkType,
                               String address,
-                              RemotingConnection remotingConnection) {
+                              RemotingConnection remotingConnection,
+                              String securityDomain) {
       System.out.println("validateUserAndRole(" + user + ", " + password + ", " + roles + ", " + checkType + ", " + address + ", " + remotingConnection.getRemoteAddress() + ")");
-      return activeMQJAASSecurityManager.validateUserAndRole(user, password, roles, checkType, address, remotingConnection);
+      return activeMQJAASSecurityManager.validateUserAndRole(user, password, roles, checkType, address, remotingConnection, securityDomain);
    }
 
    @Override

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java

@@ -589,8 +589,9 @@ public class HangConsumerTest extends ActiveMQTestBase {
                                                         SessionCallback callback,
                                                         OperationContext context,
                                                         boolean autoCreateQueue,
-                                                        Map<SimpleString, RoutingType> prefixes) throws Exception {
+                                                        Map<SimpleString, RoutingType> prefixes,
-         return new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, getConfiguration().isPersistDeliveryCountBeforeDelivery(), xa, connection, getStorageManager(), getPostOffice(), getResourceManager(), getSecurityStore(), getManagementService(), this, getConfiguration().getManagementAddress(), defaultAddress == null ? null : new SimpleString(defaultAddress), new MyCallback(callback), context, getPagingManager(), prefixes);
+                                                        String securityDomain) throws Exception {
+         return new ServerSessionImpl(name, username, password, validatedUser, minLargeMessageSize, autoCommitSends, autoCommitAcks, preAcknowledge, getConfiguration().isPersistDeliveryCountBeforeDelivery(), xa, connection, getStorageManager(), getPostOffice(), getResourceManager(), getSecurityStore(), getManagementService(), this, getConfiguration().getManagementAddress(), defaultAddress == null ? null : new SimpleString(defaultAddress), new MyCallback(callback), context, getPagingManager(), prefixes, securityDomain);
       }
    }
 

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTSecurityPerAcceptorTest.java

@@ -0,0 +1,91 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.activemq.artemis.tests.integration.mqtt.imported;
+
+import java.net.URL;
+import java.util.Arrays;
+
+import org.apache.activemq.artemis.tests.util.Wait;
+import org.fusesource.mqtt.client.BlockingConnection;
+import org.fusesource.mqtt.client.MQTT;
+import org.junit.Test;
+
+public class MQTTSecurityPerAcceptorTest extends MQTTTestSupport {
+
+   static {
+      String path = System.getProperty("java.security.auth.login.config");
+      if (path == null) {
+         URL resource = MQTTSecurityPerAcceptorTest.class.getClassLoader().getResource("login.config");
+         if (resource != null) {
+            path = resource.getFile();
+            System.setProperty("java.security.auth.login.config", path);
+         }
+      }
+   }
+
+   @Override
+   public boolean isSecurityEnabled() {
+      return true;
+   }
+
+   @Override
+   public void configureBroker() throws Exception {
+      server = createServer(true, createDefaultConfig(true).setSecurityEnabled(true));
+      server.getConfiguration().addAcceptorConfiguration("MQTT", "tcp://localhost:" + port + "?securityDomain=PropertiesLogin");
+   }
+
+   @Test(timeout = 30000)
+   public void testConnectionPositive() throws Exception {
+      internalTestConnection("first", true);
+   }
+
+   @Test(timeout = 30000)
+   public void testConnectionNegative() throws Exception {
+      internalTestConnection("fail", false);
+   }
+
+   private void internalTestConnection(String username, boolean succeed) throws Exception {
+      for (String version : Arrays.asList("3.1", "3.1.1")) {
+
+         BlockingConnection connection = null;
+         try {
+            MQTT mqtt = createMQTTConnection("test-" + version, true);
+            mqtt.setUserName(username);
+            mqtt.setPassword("secret");
+            mqtt.setConnectAttemptsMax(1);
+            mqtt.setVersion(version);
+            connection = mqtt.blockingConnection();
+            try {
+               connection.connect();
+               if (!succeed) {
+                  fail("Connection should have failed");
+               }
+            } catch (Exception e) {
+               if (succeed) {
+                  fail("Connection should have succeeded");
+               }
+            }
+            BlockingConnection finalConnection = connection;
+            if (succeed) {
+               assertTrue("Should be connected", Wait.waitFor(() -> finalConnection.isConnected(), 2000, 100));
+            }
+         } finally {
+            if (connection != null && connection.isConnected()) connection.disconnect();
+         }
+      }
+   }
+}

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/security/SecurityPerAcceptorJmsTest.java

@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.activemq.artemis.tests.integration.security;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.QueueBrowser;
+import javax.jms.Session;
+import java.lang.management.ManagementFactory;
+import java.net.URL;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.security.Role;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.ActiveMQServers;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
+import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.qpid.jms.JmsConnectionFactory;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class SecurityPerAcceptorJmsTest extends ActiveMQTestBase {
+
+   private enum Protocol {
+      CORE, AMQP, OPENWIRE
+   }
+
+   @Parameterized.Parameters(name = "protocol={0}")
+   public static Collection<Object[]> parameters() {
+      return Arrays.asList(new Object[][] {
+         {Protocol.CORE},
+         {Protocol.AMQP},
+         {Protocol.OPENWIRE}
+      });
+   }
+
+   @Parameterized.Parameter(0)
+   public Protocol protocol;
+
+   static {
+      String path = System.getProperty("java.security.auth.login.config");
+      if (path == null) {
+         URL resource = SecurityPerAcceptorJmsTest.class.getClassLoader().getResource("login.config");
+         if (resource != null) {
+            path = resource.getFile();
+            System.setProperty("java.security.auth.login.config", path);
+         }
+      }
+   }
+
+   private ConnectionFactory cf;
+   private final String URL = "tcp://127.0.0.1:61616";
+
+   @Override
+   @Before
+   public void setUp() throws Exception {
+      super.setUp();
+      switch (protocol) {
+         case CORE:
+            cf = new ActiveMQConnectionFactory(URL);
+            break;
+         case OPENWIRE:
+            cf = new org.apache.activemq.ActiveMQConnectionFactory(URL);
+            break;
+         case AMQP:
+            cf = new JmsConnectionFactory("amqp://localhost:61616");
+
+      }
+   }
+
+   @Test
+   public void testJAASSecurityManagerAuthentication() throws Exception {
+      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(createDefaultInVMConfig().setSecurityEnabled(true).addAcceptorConfiguration("netty", URL + "?securityDomain=PropertiesLogin"), ManagementFactory.getPlatformMBeanServer(), new ActiveMQJAASSecurityManager(), false));
+      server.start();
+      try (Connection c = cf.createConnection("first", "secret")) {
+         Thread.sleep(200);
+      } catch (JMSException e) {
+         Assert.fail("should not throw exception");
+      }
+   }
+
+   @Test
+   public void testJAASSecurityManagerAuthorizationNegative() throws Exception {
+      final SimpleString ADDRESS = new SimpleString("address");
+
+      ActiveMQJAASSecurityManager securityManager = new ActiveMQJAASSecurityManager();
+      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(createDefaultInVMConfig().addAcceptorConfiguration("netty", "tcp://127.0.0.1:61616?securityDomain=PropertiesLogin").setSecurityEnabled(true), ManagementFactory.getPlatformMBeanServer(), securityManager, false));
+      Set<Role> roles = new HashSet<>();
+      roles.add(new Role("programmers", false, false, false, false, false, false, false, false, false, false));
+      server.getConfiguration().putSecurityRoles("#", roles);
+      server.start();
+      server.addAddressInfo(new AddressInfo(ADDRESS, RoutingType.ANYCAST));
+      server.createQueue(new QueueConfiguration(ADDRESS).setAddress(ADDRESS).setRoutingType(RoutingType.ANYCAST));
+
+      Connection c = cf.createConnection("first", "secret");
+      Session s = c.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+      // PRODUCE
+      try {
+         MessageProducer producer = s.createProducer(s.createQueue(ADDRESS.toString()));
+         producer.send(s.createMessage());
+         Assert.fail("should throw exception here");
+      } catch (JMSException e) {
+         e.printStackTrace();
+         assertTrue(e.getMessage().contains("User: first does not have permission='SEND' on address address"));
+      }
+
+      // CONSUME
+      try {
+         MessageConsumer consumer = s.createConsumer(s.createQueue(ADDRESS.toString()));
+         Assert.fail("should throw exception here");
+      } catch (JMSException e) {
+         assertTrue(e.getMessage().contains("User: first does not have permission='CONSUME' for queue address on address address"));
+      }
+
+      // BROWSE
+      try {
+         QueueBrowser browser = s.createBrowser(s.createQueue(ADDRESS.toString()));
+         browser.getEnumeration();
+         Assert.fail("should throw exception here");
+      } catch (JMSException e) {
+         assertTrue(e.getMessage().contains("User: first does not have permission='BROWSE' for queue address on address address"));
+      }
+      c.close();
+   }
+
+   @Test
+   public void testJAASSecurityManagerAuthorizationPositive() throws Exception {
+      final String ADDRESS = "address";
+
+      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(createDefaultInVMConfig().setSecurityEnabled(true).addAcceptorConfiguration("netty", "tcp://127.0.0.1:61616?securityDomain=PropertiesLogin"), ManagementFactory.getPlatformMBeanServer(), new ActiveMQJAASSecurityManager(), false));
+      Set<Role> roles = new HashSet<>();
+      roles.add(new Role("programmers", true, true, true, true, true, true, true, true, true, true));
+      server.getConfiguration().putSecurityRoles("#", roles);
+      server.start();
+
+      Connection c = cf.createConnection("first", "secret");
+      Session s = c.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+      // PRODUCE
+      try {
+         MessageProducer producer = s.createProducer(s.createQueue(ADDRESS));
+         producer.send(s.createMessage());
+      } catch (JMSException e) {
+         Assert.fail("should not throw exception here");
+      }
+
+      // CONSUME
+      try {
+         MessageConsumer consumer = s.createConsumer(s.createQueue(ADDRESS));
+      } catch (JMSException e) {
+         Assert.fail("should not throw exception here");
+      }
+
+      // BROWSE
+      try {
+         QueueBrowser browser = s.createBrowser(s.createQueue(ADDRESS));
+         browser.getEnumeration();
+      } catch (JMSException e) {
+         Assert.fail("should not throw exception here");
+      }
+      c.close();
+   }
+}

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/security/SecurityPerAcceptorTest.java

@@ -0,0 +1,244 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.activemq.artemis.tests.integration.security;
+
+import java.lang.management.ManagementFactory;
+import java.net.URL;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.api.core.client.ClientConsumer;
+import org.apache.activemq.artemis.api.core.client.ClientProducer;
+import org.apache.activemq.artemis.api.core.client.ClientSession;
+import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
+import org.apache.activemq.artemis.api.core.client.ServerLocator;
+import org.apache.activemq.artemis.core.security.Role;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.ActiveMQServers;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
+import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class SecurityPerAcceptorTest extends ActiveMQTestBase {
+
+   static {
+      String path = System.getProperty("java.security.auth.login.config");
+      if (path == null) {
+         URL resource = SecurityPerAcceptorTest.class.getClassLoader().getResource("login.config");
+         if (resource != null) {
+            path = resource.getFile();
+            System.setProperty("java.security.auth.login.config", path);
+         }
+      }
+   }
+
+   private ServerLocator locator;
+
+   @Override
+   @Before
+   public void setUp() throws Exception {
+      super.setUp();
+
+      locator = createNettyNonHALocator();
+   }
+
+   @Test
+   public void testJAASSecurityManagerAuthentication() throws Exception {
+      ActiveMQJAASSecurityManager securityManager = new ActiveMQJAASSecurityManager();
+      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(createDefaultInVMConfig().setSecurityEnabled(true).addAcceptorConfiguration("netty", "tcp://127.0.0.1:61616?securityDomain=PropertiesLogin"), ManagementFactory.getPlatformMBeanServer(), securityManager, false));
+      server.start();
+      ClientSessionFactory cf = createSessionFactory(locator);
+
+      try {
+         ClientSession session = cf.createSession("first", "secret", false, true, true, false, 0);
+         session.close();
+      } catch (ActiveMQException e) {
+         e.printStackTrace();
+         Assert.fail("should not throw exception");
+      }
+   }
+
+   @Test
+   public void testJAASSecurityManagerAuthorizationNegative() throws Exception {
+      final SimpleString ADDRESS = new SimpleString("address");
+      final SimpleString DURABLE_QUEUE = new SimpleString("durableQueue");
+      final SimpleString NON_DURABLE_QUEUE = new SimpleString("nonDurableQueue");
+
+      ActiveMQJAASSecurityManager securityManager = new ActiveMQJAASSecurityManager();
+      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(createDefaultInVMConfig().addAcceptorConfiguration("netty", "tcp://127.0.0.1:61616?securityDomain=PropertiesLogin").setSecurityEnabled(true), ManagementFactory.getPlatformMBeanServer(), securityManager, false));
+      Set<Role> roles = new HashSet<>();
+      roles.add(new Role("programmers", false, false, false, false, false, false, false, false, false, false));
+      server.getConfiguration().putSecurityRoles("#", roles);
+      server.start();
+      server.addAddressInfo(new AddressInfo(ADDRESS, RoutingType.ANYCAST));
+      server.createQueue(new QueueConfiguration(DURABLE_QUEUE).setAddress(ADDRESS).setRoutingType(RoutingType.ANYCAST));
+      server.createQueue(new QueueConfiguration(NON_DURABLE_QUEUE).setAddress(ADDRESS).setRoutingType(RoutingType.ANYCAST).setDurable(false));
+
+      ClientSessionFactory cf = createSessionFactory(locator);
+      ClientSession session = addClientSession(cf.createSession("first", "secret", false, true, true, false, 0));
+
+      // CREATE_DURABLE_QUEUE
+      try {
+         session.createQueue(new QueueConfiguration(DURABLE_QUEUE).setAddress(ADDRESS));
+         Assert.fail("should throw exception here");
+      } catch (ActiveMQException e) {
+         assertTrue(e.getMessage().contains("User: first does not have permission='CREATE_DURABLE_QUEUE' for queue durableQueue on address address"));
+      }
+
+      // DELETE_DURABLE_QUEUE
+      try {
+         session.deleteQueue(DURABLE_QUEUE);
+         Assert.fail("should throw exception here");
+      } catch (ActiveMQException e) {
+         assertTrue(e.getMessage().contains("User: first does not have permission='DELETE_DURABLE_QUEUE' for queue durableQueue on address address"));
+      }
+
+      // CREATE_NON_DURABLE_QUEUE
+      try {
+         session.createQueue(new QueueConfiguration(NON_DURABLE_QUEUE).setAddress(ADDRESS).setDurable(false));
+         Assert.fail("should throw exception here");
+      } catch (ActiveMQException e) {
+         assertTrue(e.getMessage().contains("User: first does not have permission='CREATE_NON_DURABLE_QUEUE' for queue nonDurableQueue on address address"));
+      }
+
+      // DELETE_NON_DURABLE_QUEUE
+      try {
+         session.deleteQueue(NON_DURABLE_QUEUE);
+         Assert.fail("should throw exception here");
+      } catch (ActiveMQException e) {
+         assertTrue(e.getMessage().contains("User: first does not have permission='DELETE_NON_DURABLE_QUEUE' for queue nonDurableQueue on address address"));
+      }
+
+      // PRODUCE
+      try {
+         ClientProducer producer = session.createProducer(ADDRESS);
+         producer.send(session.createMessage(true));
+         Assert.fail("should throw exception here");
+      } catch (ActiveMQException e) {
+         assertTrue(e.getMessage().contains("User: first does not have permission='SEND' on address address"));
+      }
+
+      // CONSUME
+      try {
+         ClientConsumer consumer = session.createConsumer(DURABLE_QUEUE);
+         Assert.fail("should throw exception here");
+      } catch (ActiveMQException e) {
+         assertTrue(e.getMessage().contains("User: first does not have permission='CONSUME' for queue durableQueue on address address"));
+      }
+
+      // MANAGE
+      try {
+         ClientProducer producer = session.createProducer(server.getConfiguration().getManagementAddress());
+         producer.send(session.createMessage(true));
+         Assert.fail("should throw exception here");
+      } catch (ActiveMQException e) {
+         assertTrue(e.getMessage().contains("User: first does not have permission='MANAGE' on address activemq.management"));
+      }
+
+      // BROWSE
+      try {
+         ClientConsumer browser = session.createConsumer(DURABLE_QUEUE, true);
+         Assert.fail("should throw exception here");
+      } catch (ActiveMQException e) {
+         assertTrue(e.getMessage().contains("User: first does not have permission='BROWSE' for queue durableQueue on address address"));
+      }
+   }
+
+   @Test
+   public void testJAASSecurityManagerAuthorizationPositive() throws Exception {
+      final SimpleString ADDRESS = new SimpleString("address");
+      final SimpleString DURABLE_QUEUE = new SimpleString("durableQueue");
+      final SimpleString NON_DURABLE_QUEUE = new SimpleString("nonDurableQueue");
+
+      ActiveMQJAASSecurityManager securityManager = new ActiveMQJAASSecurityManager();
+      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(createDefaultInVMConfig().setSecurityEnabled(true).addAcceptorConfiguration("netty", "tcp://127.0.0.1:61616?securityDomain=PropertiesLogin"), ManagementFactory.getPlatformMBeanServer(), securityManager, false));
+      Set<Role> roles = new HashSet<>();
+      roles.add(new Role("programmers", true, true, true, true, true, true, true, true, true, true));
+      server.getConfiguration().putSecurityRoles("#", roles);
+      server.start();
+
+      ClientSessionFactory cf = createSessionFactory(locator);
+      ClientSession session = addClientSession(cf.createSession("first", "secret", false, true, true, false, 0));
+
+      // CREATE_DURABLE_QUEUE
+      try {
+         session.createQueue(new QueueConfiguration(DURABLE_QUEUE).setAddress(ADDRESS));
+      } catch (ActiveMQException e) {
+         Assert.fail("should not throw exception here");
+      }
+
+      // DELETE_DURABLE_QUEUE
+      try {
+         session.deleteQueue(DURABLE_QUEUE);
+      } catch (ActiveMQException e) {
+         Assert.fail("should not throw exception here");
+      }
+
+      // CREATE_NON_DURABLE_QUEUE
+      try {
+         session.createQueue(new QueueConfiguration(NON_DURABLE_QUEUE).setAddress(ADDRESS).setDurable(false));
+      } catch (ActiveMQException e) {
+         Assert.fail("should not throw exception here");
+      }
+
+      // DELETE_NON_DURABLE_QUEUE
+      try {
+         session.deleteQueue(NON_DURABLE_QUEUE);
+      } catch (ActiveMQException e) {
+         Assert.fail("should not throw exception here");
+      }
+
+      session.createQueue(new QueueConfiguration(DURABLE_QUEUE).setAddress(ADDRESS));
+
+      // PRODUCE
+      try {
+         ClientProducer producer = session.createProducer(ADDRESS);
+         producer.send(session.createMessage(true));
+      } catch (ActiveMQException e) {
+         Assert.fail("should not throw exception here");
+      }
+
+      // CONSUME
+      try {
+         session.createConsumer(DURABLE_QUEUE);
+      } catch (ActiveMQException e) {
+         Assert.fail("should not throw exception here");
+      }
+
+      // MANAGE
+      try {
+         ClientProducer producer = session.createProducer(server.getConfiguration().getManagementAddress());
+         producer.send(session.createMessage(true));
+      } catch (ActiveMQException e) {
+         Assert.fail("should not throw exception here");
+      }
+
+      // BROWSE
+      try {
+         session.createConsumer(DURABLE_QUEUE, true);
+      } catch (ActiveMQException e) {
+         Assert.fail("should not throw exception here");
+      }
+   }
+}

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/security/SecurityTest.java

@@ -62,6 +62,7 @@ import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager
 import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager;
 import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager2;
 import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager3;
+import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager4;
 import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
 import org.apache.activemq.artemis.tests.util.CreateMessage;
 import org.apache.activemq.command.ActiveMQQueue;
@@ -2202,6 +2203,142 @@ public class SecurityTest extends ActiveMQTestBase {
       }
    }
 
+   @Test
+   public void testCustomSecurityManager4() throws Exception {
+      final Configuration configuration = createDefaultInVMConfig().setSecurityEnabled(true);
+      final ActiveMQSecurityManager customSecurityManager = new ActiveMQSecurityManager4() {
+         @Override
+         public boolean validateUser(final String username, final String password) {
+            fail("Unexpected call to overridden method");
+            return false;
+         }
+
+         @Override
+         public String validateUser(final String username,
+                                    final String password,
+                                    final RemotingConnection remotingConnection,
+                                    final String securityDomain) {
+            if ((username.equals("foo") || username.equals("bar") || username.equals("all")) && password.equals("frobnicate")) {
+               return username;
+            } else {
+               return null;
+            }
+         }
+
+         @Override
+         public boolean validateUserAndRole(final String username,
+                                            final String password,
+                                            final Set<Role> requiredRoles,
+                                            final CheckType checkType) {
+
+            fail("Unexpected call to overridden method");
+            return false;
+         }
+
+         @Override
+         public String validateUserAndRole(final String username,
+                                           final String password,
+                                           final Set<Role> requiredRoles,
+                                           final CheckType checkType,
+                                           final String address,
+                                           final RemotingConnection connection,
+                                           final String securityDomain) {
+
+            if (!(connection.getTransportConnection() instanceof InVMConnection)) {
+               return null;
+            }
+
+            if ((username.equals("foo") || username.equals("bar") || username.equals("all")) && password.equals("frobnicate")) {
+
+               if (username.equals("all")) {
+                  return username;
+               } else if (username.equals("foo")) {
+                  if (address.equals("test.queue") && checkType == CheckType.CONSUME)
+                     return username;
+                  else
+                     return null;
+               } else if (username.equals("bar")) {
+                  if (address.equals("test.queue") && checkType == CheckType.SEND)
+                     return username;
+                  else
+                     return null;
+               } else {
+                  return null;
+               }
+            } else {
+               return null;
+            }
+         }
+      };
+      final ActiveMQServer server = addServer(new ActiveMQServerImpl(configuration, customSecurityManager));
+      server.start();
+
+      final ServerLocator locator = createInVMNonHALocator();
+      locator.setBlockOnNonDurableSend(true).setBlockOnDurableSend(true);
+      final ClientSessionFactory factory = createSessionFactory(locator);
+      ClientSession adminSession = factory.createSession("all", "frobnicate", false, true, true, false, -1);
+
+      final String queueName = "test.queue";
+      adminSession.createQueue(new QueueConfiguration(queueName).setDurable(false));
+
+      final String otherQueueName = "other.queue";
+      adminSession.createQueue(new QueueConfiguration(otherQueueName).setDurable(false));
+
+      // Wrong user name
+      try {
+         factory.createSession("baz", "frobnicate", false, true, true, false, -1);
+         Assert.fail("should throw exception");
+      } catch (ActiveMQSecurityException se) {
+         //ok
+      } catch (ActiveMQException e) {
+         fail("Invalid Exception type:" + e.getType());
+      }
+
+      // Wrong password
+      try {
+         factory.createSession("foo", "xxx", false, true, true, false, -1);
+         Assert.fail("should throw exception");
+      } catch (ActiveMQSecurityException se) {
+         //ok
+      } catch (ActiveMQException e) {
+         fail("Invalid Exception type:" + e.getType());
+      }
+
+      // Correct user and password, wrong queue for sending
+      try {
+         final ClientSession session = factory.createSession("foo", "frobnicate", false, true, true, false, -1);
+         checkUserReceiveNoSend(otherQueueName, session, adminSession);
+         Assert.fail("should throw exception");
+      } catch (ActiveMQSecurityException se) {
+         //ok
+      } catch (ActiveMQException e) {
+         fail("Invalid Exception type:" + e.getType());
+      }
+
+      // Correct user and password, wrong queue for receiving
+      try {
+         final ClientSession session = factory.createSession("foo", "frobnicate", false, true, true, false, -1);
+         checkUserReceiveNoSend(otherQueueName, session, adminSession);
+         Assert.fail("should throw exception");
+      } catch (ActiveMQSecurityException se) {
+         //ok
+      } catch (ActiveMQException e) {
+         fail("Invalid Exception type:" + e.getType());
+      }
+
+      // Correct user and password, allowed to send but not receive
+      {
+         final ClientSession session = factory.createSession("foo", "frobnicate", false, true, true, false, -1);
+         checkUserReceiveNoSend(queueName, session, adminSession);
+      }
+
+      // Correct user and password, allowed to receive but not send
+      {
+         final ClientSession session = factory.createSession("bar", "frobnicate", false, true, true, false, -1);
+         checkUserSendNoReceive(queueName, session);
+      }
+   }
+
    // Check the user connection has both send and receive permissions on the queue
    private void checkUserSendAndReceive(final String genericQueueName,
                                         final ClientSession connection) throws Exception {

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/server/SecureConfigurationTest.java

@@ -286,12 +286,12 @@ public class SecureConfigurationTest extends ActiveMQTestBase {
          try (Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
             D destination = destinationSupplier.create(session);
             MessageConsumer messageConsumer = consumerSupplier.create(destination, session);
-            messageConsumer.receive(1000);
+            Assert.assertNull(messageConsumer.receiveNoWait());
 
             TextMessage messageToSend = session.createTextMessage(message);
             session.createProducer(destination).send(messageToSend);
 
-            TextMessage received = (TextMessage) messageConsumer.receive(1000);
+            TextMessage received = (TextMessage) messageConsumer.receive(100);
             messageRecieved = received != null ? received.getText() : null;
          }
       } catch (JMSException | JMSRuntimeException e) {

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompWithClientIdValidationTest.java

@@ -50,9 +50,9 @@ public class StompWithClientIdValidationTest extends StompTestBase {
 
       ActiveMQJAASSecurityManager securityManager = new ActiveMQJAASSecurityManager(InVMLoginModule.class.getName(), new SecurityConfiguration()) {
          @Override
-         public String validateUser(String user, String password, RemotingConnection remotingConnection) {
+         public String validateUser(String user, String password, RemotingConnection remotingConnection, String securityDomain) {
 
-            String validatedUser = super.validateUser(user, password, remotingConnection);
+            String validatedUser = super.validateUser(user, password, remotingConnection, securityDomain);
             if (validatedUser == null) {
                return null;
             }

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompWithSecurityPerAcceptorTest.java

@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.activemq.artemis.tests.integration.stomp;
+
+import java.lang.management.ManagementFactory;
+import java.net.URI;
+import java.net.URL;
+
+import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.protocol.stomp.Stomp;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.ActiveMQServers;
+import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
+import org.apache.activemq.artemis.tests.integration.stomp.util.ClientStompFrame;
+import org.apache.activemq.artemis.tests.integration.stomp.util.StompClientConnection;
+import org.apache.activemq.artemis.tests.integration.stomp.util.StompClientConnectionFactory;
+import org.junit.Before;
+import org.junit.Test;
+
+public class StompWithSecurityPerAcceptorTest extends StompTestBase {
+
+   static {
+      String path = System.getProperty("java.security.auth.login.config");
+      if (path == null) {
+         URL resource = StompWithSecurityPerAcceptorTest.class.getClassLoader().getResource("login.config");
+         if (resource != null) {
+            path = resource.getFile();
+            System.setProperty("java.security.auth.login.config", path);
+         }
+      }
+   }
+
+   @Override
+   public boolean isSecurityEnabled() {
+      return true;
+   }
+
+   @Override
+   @Before
+   public void setUp() throws Exception {
+      uri = new URI(scheme + "://" + hostname + ":" + port);
+
+      server = createServer();
+      server.start();
+
+      waitForServerToStart(server);
+   }
+
+   @Override
+   protected ActiveMQServer createServer() throws Exception {
+      Configuration config = createBasicConfig()
+         .setSecurityEnabled(isSecurityEnabled())
+         .setPersistenceEnabled(isPersistenceEnabled())
+         .addAcceptorConfiguration("stomp", "tcp://localhost:61613?securityDomain=PropertiesLogin");
+
+      server = addServer(ActiveMQServers.newActiveMQServer(config, ManagementFactory.getPlatformMBeanServer(), new ActiveMQJAASSecurityManager()));
+      return server;
+   }
+
+   @Test
+   public void testSecurityPerAcceptorPositive() throws Exception {
+      StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri);
+      ClientStompFrame frame = conn.connect("first", "secret");
+      assertTrue(frame.getCommand().equals(Stomp.Responses.CONNECTED));
+   }
+
+   @Test
+   public void testSecurityPerAcceptorNegative() throws Exception {
+      StompClientConnection conn = StompClientConnectionFactory.createClientConnection(uri);
+      ClientStompFrame frame = conn.connect("fail", "secret");
+      assertTrue(frame.getCommand().equals(Stomp.Responses.ERROR));
+   }
+}