ARTEMIS-3012 incorrect fallback consumer authorization

The fallback consumer authorization implemented in ARTEMIS-592 needs to
check for an *exact* security-settings match otherwise in certain
configurations a more general and more permissive setting might
be used instead of the intended more specific and more restrictive
setting.

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

@@ -541,8 +541,17 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
       try {
          securityCheck(address, unPrefixedQueueName, browseOnly ? CheckType.BROWSE : CheckType.CONSUME, this);
       } catch (Exception e) {
-         // this is here for backwards compatibility with the pre-FQQN syntax from ARTEMIS-592
-         securityCheck(address.concat(".").concat(unPrefixedQueueName), queueName, browseOnly ? CheckType.BROWSE : CheckType.CONSUME, this);
+         /*
+          * This is here for backwards compatibility with the pre-FQQN syntax from ARTEMIS-592.
+          * We only want to do this check if an exact match exists in the security-settings.
+          * This code is deprecated and should be removed at the release of the next major version.
+          */
+         SimpleString exactMatch = address.concat(".").concat(unPrefixedQueueName);
+         if (server.getSecurityRepository().containsExactMatch(exactMatch.toString())) {
+            securityCheck(exactMatch, unPrefixedQueueName, browseOnly ? CheckType.BROWSE : CheckType.CONSUME, this);
+         } else {
+            throw e;
+         }
       }
 
       Filter filter = FilterImpl.createFilter(filterString);

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

@@ -595,6 +595,37 @@ public class SecurityTest extends ActiveMQTestBase {
       }
    }
 
+   @Test
+   public void testFallbackConsumerAuthorization() throws Exception {
+      final SimpleString ADDRESS = new SimpleString("a.c.b");
+      final SimpleString QUEUE = new SimpleString("a.c.b");
+
+      ActiveMQJAASSecurityManager securityManager = new ActiveMQJAASSecurityManager("PropertiesLogin");
+      ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(createDefaultInVMConfig().setSecurityEnabled(true), ManagementFactory.getPlatformMBeanServer(), securityManager, false));
+
+      Set<Role> aRoles = new HashSet<>();
+      aRoles.add(new Role("xyz", true, true, true, true, true, true, true, true, true, true));
+      server.getConfiguration().putSecurityRoles("a.*.b", aRoles);
+
+      Set<Role> bRoles = new HashSet<>();
+      bRoles.add(new Role("amq", true, true, true, true, true, true, true, true, true, true));
+      server.getConfiguration().putSecurityRoles("#", bRoles);
+
+      server.start();
+      server.addAddressInfo(new AddressInfo(ADDRESS, RoutingType.ANYCAST));
+      server.createQueue(new QueueConfiguration(QUEUE).setAddress(ADDRESS).setRoutingType(RoutingType.ANYCAST));
+
+      ClientSessionFactory cf = createSessionFactory(locator);
+      ClientSession session = addClientSession(cf.createSession("x", "x", false, true, true, false, 0));
+
+      try {
+         session.createConsumer(QUEUE);
+         Assert.fail("should throw exception here");
+      } catch (ActiveMQException e) {
+         assertTrue(e instanceof ActiveMQSecurityException);
+      }
+   }
+
    @Test
    public void testJAASSecurityManagerFQQNAuthorizationWithJMS() throws Exception {
       final SimpleString ADDRESS = new SimpleString("address");

tests/integration-tests/src/test/resources/roles.properties

@@ -20,3 +20,4 @@ accounting=second
 employees=first,second
 a=a
 b=b
+amq=x

tests/integration-tests/src/test/resources/users.properties

@@ -19,3 +19,4 @@ first=secret
 second=password
 a=a
 b=b
+x=x