https://issues.apache.org/jira/browse/AMQ-5470

Allow for early SASL authentication and failure if credentials not
valid.

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java

@@ -19,6 +19,8 @@ package org.apache.activemq.transport.amqp;
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.nio.ByteBuffer;
+import java.security.Principal;
+import java.security.cert.X509Certificate;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -27,6 +29,7 @@ import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 import javax.jms.Destination;
@@ -65,6 +68,8 @@ import org.apache.activemq.command.SessionInfo;
 import org.apache.activemq.command.ShutdownInfo;
 import org.apache.activemq.command.SubscriptionInfo;
 import org.apache.activemq.command.TransactionInfo;
+import org.apache.activemq.security.AuthenticationBroker;
+import org.apache.activemq.security.SecurityContext;
 import org.apache.activemq.selector.SelectorParser;
 import org.apache.activemq.store.PersistenceAdapterSupport;
 import org.apache.activemq.transport.amqp.message.AMQPNativeInboundTransformer;
@@ -139,6 +144,7 @@ class AmqpProtocolConverter implements IAmqpProtocolConverter {
     private final AmqpTransport amqpTransport;
     private final AmqpWireFormat amqpWireFormat;
     private final BrokerService brokerService;
+    private AuthenticationBroker authenticator;
 
     protected int prefetch;
     protected int producerCredit;
@@ -310,14 +316,22 @@ class AmqpProtocolConverter implements IAmqpProtocolConverter {
                             if (parts.length > 1) {
                                 connectionInfo.setPassword(parts[1].utf8().toString());
                             }
-                            // We can't really auth at this point since we don't
+
-                            // know the client id yet.. :(
+                            if (tryAuthenticate(connectionInfo, amqpTransport.getPeerCertificates())) {
                                 sasl.done(Sasl.SaslOutcome.PN_SASL_OK);
+                            } else {
+                                sasl.done(Sasl.SaslOutcome.PN_SASL_AUTH);
+                            }
+
                             amqpTransport.getWireFormat().resetMagicRead();
                             sasl = null;
                             LOG.debug("SASL [PLAIN] Handshake complete.");
                         } else if ("ANONYMOUS".equals(sasl.getRemoteMechanisms()[0])) {
+                            if (tryAuthenticate(connectionInfo, amqpTransport.getPeerCertificates())) {
                                 sasl.done(Sasl.SaslOutcome.PN_SASL_OK);
+                            } else {
+                                sasl.done(Sasl.SaslOutcome.PN_SASL_AUTH);
+                            }
                             amqpTransport.getWireFormat().resetMagicRead();
                             sasl = null;
                             LOG.debug("SASL [ANONYMOUS] Handshake complete.");
@@ -1690,4 +1704,46 @@ class AmqpProtocolConverter implements IAmqpProtocolConverter {
 
         return subscriptions;
     }
+
+    public boolean tryAuthenticate(ConnectionInfo info, X509Certificate[] peerCertificates) {
+        try {
+            if (getAuthenticator().authenticate(info.getUserName(), info.getPassword(), peerCertificates) != null) {
+                return true;
+            }
+
+            return false;
+        } catch (Throwable error) {
+            return false;
+        }
+    }
+
+    private AuthenticationBroker getAuthenticator() {
+        if (authenticator == null) {
+            try {
+                authenticator = (AuthenticationBroker) brokerService.getBroker().getAdaptor(AuthenticationBroker.class);
+            } catch (Exception e) {
+                LOG.debug("Failed to lookup AuthenticationBroker from Broker, will use a default Noop version.");
+            }
+
+            if (authenticator == null) {
+                authenticator = new DefaultAuthenticationBroker();
+            }
+        }
+
+        return authenticator;
+    }
+
+    private class DefaultAuthenticationBroker implements AuthenticationBroker {
+
+        @Override
+        public SecurityContext authenticate(String username, String password, X509Certificate[] peerCertificates) throws SecurityException {
+            return new SecurityContext(username) {
+
+                @Override
+                public Set<Principal> getPrincipals() {
+                    return null;
+                }
+            };
+        }
+    }
 }

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientSimpleAuthTest.java

@@ -69,8 +69,6 @@ public class JMSClientSimpleAuthTest {
         try {
             Connection connection = JMSClientContext.INSTANCE.createConnection(amqpURI, "", "");
             connection.start();
-            Thread.sleep(500);
-            connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             fail("Expected JMSException");
         } catch (JMSSecurityException ex) {
             LOG.debug("Failed to authenticate connection with no user / password.");
@@ -91,8 +89,6 @@ public class JMSClientSimpleAuthTest {
         try {
             Connection connection = JMSClientContext.INSTANCE.createConnection(amqpURI, "nosuchuser", "blah");
             connection.start();
-            Thread.sleep(500);
-            connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             fail("Expected JMSException");
         } catch (JMSSecurityException ex) {
             LOG.debug("Failed to authenticate connection with no user / password.");
@@ -113,8 +109,6 @@ public class JMSClientSimpleAuthTest {
         try {
             Connection connection = JMSClientContext.INSTANCE.createConnection(amqpURI, "user", "wrongPassword");
             connection.start();
-            Thread.sleep(500);
-            connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             fail("Expected JMSException");
         } catch (JMSSecurityException ex) {
             LOG.debug("Failed to authenticate connection with no user / password.");
@@ -130,6 +124,33 @@ public class JMSClientSimpleAuthTest {
         }
     }
 
+    @Test(timeout = 30000)
+    public void testRepeatedWrongPasswordAttempts() throws Exception {
+        for (int i = 0; i < 25; ++i) {
+            Connection connection = null;
+            try {
+                connection = JMSClientContext.INSTANCE.createConnection(amqpURI, "user", "wrongPassword");
+                connection.start();
+                fail("Expected JMSException");
+            } catch (JMSSecurityException ex) {
+                LOG.debug("Failed to authenticate connection with no user / password.");
+            } catch (JMSException e) {
+                Exception linkedException = e.getLinkedException();
+                if (linkedException != null && linkedException instanceof ConnectionClosedException) {
+                    ConnectionClosedException cce = (ConnectionClosedException) linkedException;
+                    assertEquals("Error{condition=unauthorized-access,description=User name [user] or password is invalid.}", cce.getRemoteError().toString());
+                } else {
+                    LOG.error("Unexpected Exception", e);
+                    fail("Unexpected exception: " + e.getMessage());
+                }
+            } finally {
+                if (connection != null) {
+                    connection.close();
+                }
+            }
+        }
+    }
+
     @Test(timeout = 30000)
     public void testSendReceive() throws Exception {
         Connection connection = JMSClientContext.INSTANCE.createConnection(amqpURI, "user", "userPassword");
@@ -205,4 +226,3 @@ public class JMSClientSimpleAuthTest {
         brokerService.waitUntilStarted();
     }
 }
-

activemq-broker/src/main/java/org/apache/activemq/security/AbstractAuthenticationBroker.java

@@ -24,7 +24,7 @@ import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ConnectionInfo;
 
-public class AbstractAuthenticationBroker extends BrokerFilter {
+public abstract class AbstractAuthenticationBroker extends BrokerFilter implements AuthenticationBroker {
 
     protected final CopyOnWriteArrayList<SecurityContext> securityContexts =
         new CopyOnWriteArrayList<SecurityContext>();
@@ -51,10 +51,6 @@ public class AbstractAuthenticationBroker extends BrokerFilter {
         }
     }
 
-    /**
-     * Previously logged in users may no longer have the same access anymore.
-     * Refresh all the logged into users.
-     */
     public void refresh() {
         for (SecurityContext sc : securityContexts) {
             sc.getAuthorizedReadDests().clear();

activemq-broker/src/main/java/org/apache/activemq/security/AuthenticationBroker.java

@@ -0,0 +1,42 @@
+/**
+ * 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.security;
+
+import java.security.cert.X509Certificate;
+
+/**
+ * Base for all broker plugins that wish to provide connection authentication services
+ */
+public interface AuthenticationBroker {
+
+    /**
+     * Authenticate the given user using the mechanism provided by this service.
+     *
+     * @param username
+     *        the given user name to authenticate, null indicates an anonymous user.
+     * @param password
+     *        the given password for the user to authenticate.
+     * @param peerCertificates
+     *        for an SSL channel the certificates from remote peer.
+     *
+     * @return a new SecurityContext for the authenticated user.
+     *
+     * @throws SecurityException if the user cannot be authenticated.
+     */
+    SecurityContext authenticate(String username, String password, X509Certificate[] peerCertificates) throws SecurityException;
+
+}

activemq-broker/src/main/java/org/apache/activemq/security/DefaultAuthorizationMap.java

@@ -16,10 +16,6 @@
  */
 package org.apache.activemq.security;
 
-import org.apache.activemq.command.ActiveMQDestination;
-import org.apache.activemq.filter.DestinationMap;
-import org.apache.activemq.filter.DestinationMapEntry;
-
 import java.lang.reflect.Constructor;
 import java.lang.reflect.Method;
 import java.security.Principal;
@@ -29,6 +25,10 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
 
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.filter.DestinationMap;
+import org.apache.activemq.filter.DestinationMapEntry;
+
 /**
  * Represents a destination based configuration of policies so that individual
  * destinations or wildcard hierarchies of destinations can be configured using
@@ -64,6 +64,7 @@ public class DefaultAuthorizationMap extends DestinationMap implements Authoriza
         return this.tempDestinationAuthorizationEntry;
     }
 
+    @Override
     public Set<Object> getTempDestinationAdminACLs() {
         if (tempDestinationAuthorizationEntry != null) {
             Set<Object> answer = new WildcardAwareSet<Object>();
@@ -74,6 +75,7 @@ public class DefaultAuthorizationMap extends DestinationMap implements Authoriza
         }
     }
 
+    @Override
     public Set<Object> getTempDestinationReadACLs() {
         if (tempDestinationAuthorizationEntry != null) {
             Set<Object> answer = new WildcardAwareSet<Object>();
@@ -84,6 +86,7 @@ public class DefaultAuthorizationMap extends DestinationMap implements Authoriza
         }
     }
 
+    @Override
     public Set<Object> getTempDestinationWriteACLs() {
         if (tempDestinationAuthorizationEntry != null) {
             Set<Object> answer = new WildcardAwareSet<Object>();
@@ -94,6 +97,7 @@ public class DefaultAuthorizationMap extends DestinationMap implements Authoriza
         }
     }
 
+    @Override
     public Set<Object> getAdminACLs(ActiveMQDestination destination) {
         Set<AuthorizationEntry> entries = getAllEntries(destination);
         Set<Object> answer = new WildcardAwareSet<Object>();
@@ -106,6 +110,7 @@ public class DefaultAuthorizationMap extends DestinationMap implements Authoriza
         return answer;
     }
 
+    @Override
     public Set<Object> getReadACLs(ActiveMQDestination destination) {
         Set<AuthorizationEntry> entries = getAllEntries(destination);
         Set<Object> answer = new WildcardAwareSet<Object>();
@@ -118,6 +123,7 @@ public class DefaultAuthorizationMap extends DestinationMap implements Authoriza
         return answer;
     }
 
+    @Override
     public Set<Object> getWriteACLs(ActiveMQDestination destination) {
         Set<AuthorizationEntry> entries = getAllEntries(destination);
         Set<Object> answer = new WildcardAwareSet<Object>();
@@ -150,7 +156,7 @@ public class DefaultAuthorizationMap extends DestinationMap implements Authoriza
      *         matching values.
      */
     @Override
-    @SuppressWarnings({ "rawtypes", "unchecked" })
+    @SuppressWarnings("rawtypes")
     public synchronized Set get(ActiveMQDestination key) {
         if (key.isComposite()) {
             ActiveMQDestination[] destinations = key.getCompositeDestinations();
@@ -184,6 +190,7 @@ public class DefaultAuthorizationMap extends DestinationMap implements Authoriza
         this.defaultEntry = defaultEntry;
     }
 
+    @Override
     @SuppressWarnings("rawtypes")
     protected Class<? extends DestinationMapEntry> getEntryClass() {
         return AuthorizationEntry.class;

activemq-broker/src/main/java/org/apache/activemq/security/JaasAuthenticationBroker.java

@@ -17,6 +17,7 @@
 package org.apache.activemq.security;
 
 import java.security.Principal;
+import java.security.cert.X509Certificate;
 import java.util.Set;
 
 import javax.security.auth.Subject;
@@ -58,32 +59,36 @@ public class JaasAuthenticationBroker extends AbstractAuthenticationBroker {
 
     @Override
     public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
-
         if (context.getSecurityContext() == null) {
-            // Set the TCCL since it seems JAAS needs it to find the login
+            // Set the TCCL since it seems JAAS needs it to find the login module classes.
-            // module classes.
             ClassLoader original = Thread.currentThread().getContextClassLoader();
             Thread.currentThread().setContextClassLoader(JaasAuthenticationBroker.class.getClassLoader());
-            try {
-                // Do the login.
-                try {
-                    JassCredentialCallbackHandler callback = new JassCredentialCallbackHandler(info
-                        .getUserName(), info.getPassword());
-                    LoginContext lc = new LoginContext(jassConfiguration, callback);
-                    lc.login();
-                    Subject subject = lc.getSubject();
 
-                    SecurityContext s = new JaasSecurityContext(info.getUserName(), subject);
+            try {
+                SecurityContext s = authenticate(info.getUserName(), info.getPassword(), null);
                 context.setSecurityContext(s);
                 securityContexts.add(s);
-                } catch (Exception e) {
-                    throw (SecurityException)new SecurityException("User name [" + info.getUserName() + "] or password is invalid.")
-                        .initCause(e);
-                }
             } finally {
                 Thread.currentThread().setContextClassLoader(original);
             }
         }
         super.addConnection(context, info);
     }
+
+    @Override
+    public SecurityContext authenticate(String username, String password, X509Certificate[] certificates) throws SecurityException {
+        SecurityContext result = null;
+        JassCredentialCallbackHandler callback = new JassCredentialCallbackHandler(username, password);
+        try {
+            LoginContext lc = new LoginContext(jassConfiguration, callback);
+            lc.login();
+            Subject subject = lc.getSubject();
+
+            result = new JaasSecurityContext(username, subject);
+        } catch (Exception ex) {
+            throw new SecurityException("User name [" + username + "] or password is invalid.", ex);
+        }
+
+        return result;
+    }
 }

activemq-broker/src/main/java/org/apache/activemq/security/JaasCertificateAuthenticationBroker.java

@@ -37,10 +37,8 @@ import org.apache.activemq.jaas.UserPrincipal;
  * grant JAAS access to incoming connections' SSL certificate chains. NOTE:
  * There is a chance that the incoming connection does not have a valid
  * certificate (has null).
- * 
- * @author sepandm@gmail.com (Sepand)
  */
-public class JaasCertificateAuthenticationBroker extends BrokerFilter {
+public class JaasCertificateAuthenticationBroker extends BrokerFilter implements AuthenticationBroker {
     private final String jaasConfiguration;
 
     /**
@@ -67,6 +65,7 @@ public class JaasCertificateAuthenticationBroker extends BrokerFilter {
      * @param info The ConnectionInfo Command representing the incoming
      *                connection.
      */
+    @Override
     public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
 
         if (context.getSecurityContext() == null) {
@@ -79,9 +78,29 @@ public class JaasCertificateAuthenticationBroker extends BrokerFilter {
             ClassLoader original = Thread.currentThread().getContextClassLoader();
             Thread.currentThread().setContextClassLoader(JaasAuthenticationBroker.class.getClassLoader());
             try {
-                // Do the login.
+                SecurityContext s = authenticate(info.getUserName(), info.getPassword(), (X509Certificate[]) info.getTransportContext());
+                context.setSecurityContext(s);
+            } finally {
+                Thread.currentThread().setContextClassLoader(original);
+            }
+        }
+        super.addConnection(context, info);
+    }
+
+    /**
+     * Overriding removeConnection to make sure the security context is cleaned.
+     */
+    @Override
+    public void removeConnection(ConnectionContext context, ConnectionInfo info, Throwable error) throws Exception {
+        super.removeConnection(context, info, error);
+
+        context.setSecurityContext(null);
+    }
+
+    @Override
+    public SecurityContext authenticate(String username, String password, X509Certificate[] peerCertificates) throws SecurityException {
         try {
-                    CallbackHandler callback = new JaasCertificateCallbackHandler((X509Certificate[])info.getTransportContext());
+            CallbackHandler callback = new JaasCertificateCallbackHandler(peerCertificates);
             LoginContext lc = new LoginContext(jaasConfiguration, callback);
             lc.login();
             Subject subject = lc.getSubject();
@@ -94,24 +113,10 @@ public class JaasCertificateAuthenticationBroker extends BrokerFilter {
                     break;
                 }
             }
-                    SecurityContext s = new JaasCertificateSecurityContext(dnName, subject, (X509Certificate[])info.getTransportContext());
+
-                    context.setSecurityContext(s);
+            return new JaasCertificateSecurityContext(dnName, subject, peerCertificates);
         } catch (Exception e) {
-                    throw new SecurityException("User name [" + info.getUserName() + "] or password is invalid. " + e.getMessage(), e);
+            throw new SecurityException("User name [" + username + "] or password is invalid. " + e.getMessage(), e);
         }
-            } finally {
-                Thread.currentThread().setContextClassLoader(original);
-            }
-        }
-        super.addConnection(context, info);
-    }
-
-    /**
-     * Overriding removeConnection to make sure the security context is cleaned.
-     */
-    public void removeConnection(ConnectionContext context, ConnectionInfo info, Throwable error) throws Exception {
-        super.removeConnection(context, info, error);
-
-        context.setSecurityContext(null);
     }
 }

activemq-broker/src/main/java/org/apache/activemq/security/JaasDualAuthenticationBroker.java

@@ -17,6 +17,8 @@
 
 package org.apache.activemq.security;
 
+import java.security.cert.X509Certificate;
+
 import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.BrokerFilter;
 import org.apache.activemq.broker.ConnectionContext;
@@ -56,7 +58,7 @@ import org.apache.activemq.transport.tcp.SslTransportServer;
  * };
  * </pre>
  */
-public class JaasDualAuthenticationBroker extends BrokerFilter {
+public class JaasDualAuthenticationBroker extends BrokerFilter implements AuthenticationBroker {
     private final JaasCertificateAuthenticationBroker sslBroker;
     private final JaasAuthenticationBroker nonSslBroker;
 
@@ -87,13 +89,11 @@ public class JaasDualAuthenticationBroker extends BrokerFilter {
     @Override
     public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
         if (context.getSecurityContext() == null) {
-            boolean isSSL;
+            boolean isSSL = false;
             Connector connector = context.getConnector();
             if (connector instanceof TransportConnector) {
                 TransportConnector transportConnector = (TransportConnector) connector;
                 isSSL = transportConnector.getServer().isSslServer();
-            } else {
-                isSSL = false;
             }
 
             if (isSSL) {
@@ -134,4 +134,13 @@ public class JaasDualAuthenticationBroker extends BrokerFilter {
 
         super.removeDestination(context, destination, timeout);
     }
+
+    @Override
+    public SecurityContext authenticate(String username, String password, X509Certificate[] peerCertificates) throws SecurityException {
+        if (peerCertificates != null) {
+            return this.sslBroker.authenticate(username, password, peerCertificates);
+        } else {
+            return this.nonSslBroker.authenticate(username, password, peerCertificates);
+        }
+    }
 }

activemq-broker/src/main/java/org/apache/activemq/security/SimpleAuthenticationBroker.java

@@ -17,6 +17,7 @@
 package org.apache.activemq.security;
 
 import java.security.Principal;
+import java.security.cert.X509Certificate;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
@@ -65,13 +66,30 @@ public class SimpleAuthenticationBroker extends AbstractAuthenticationBroker {
 
     @Override
     public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
+        SecurityContext securityContext = context.getSecurityContext();
+        if (securityContext == null) {
+            securityContext = authenticate(info.getUserName(), info.getPassword(), null);
+            context.setSecurityContext(securityContext);
+            securityContexts.add(securityContext);
+        }
+
+        try {
+            super.addConnection(context, info);
+        } catch (Exception e) {
+            securityContexts.remove(securityContext);
+            context.setSecurityContext(null);
+            throw e;
+        }
+    }
+
+    @Override
+    public SecurityContext authenticate(String username, String password, X509Certificate[] certificates) throws SecurityException {
+        SecurityContext securityContext = null;
 
-        SecurityContext s = context.getSecurityContext();
-        if (s == null) {
         // Check the username and password.
-            if (anonymousAccessAllowed && info.getUserName() == null && info.getPassword() == null) {
+        if (anonymousAccessAllowed && username == null && password == null) {
-                info.setUserName(anonymousUser);
+            username = anonymousUser;
-                s = new SecurityContext(info.getUserName()) {
+            securityContext = new SecurityContext(username) {
                 @Override
                 public Set<Principal> getPrincipals() {
                     Set<Principal> groups = new HashSet<Principal>();
@@ -80,14 +98,13 @@ public class SimpleAuthenticationBroker extends AbstractAuthenticationBroker {
                 }
             };
         } else {
-                String pw = userPasswords.get(info.getUserName());
+            String pw = userPasswords.get(username);
-                if (pw == null || !pw.equals(info.getPassword())) {
+            if (pw == null || !pw.equals(password)) {
-                    throw new SecurityException(
+                throw new SecurityException("User name [" + username + "] or password is invalid.");
-                            "User name [" + info.getUserName() + "] or password is invalid.");
             }
 
-                final Set<Principal> groups = userGroups.get(info.getUserName());
+            final Set<Principal> groups = userGroups.get(username);
-                s = new SecurityContext(info.getUserName()) {
+            securityContext = new SecurityContext(username) {
                 @Override
                 public Set<Principal> getPrincipals() {
                     return groups;
@@ -95,16 +112,6 @@ public class SimpleAuthenticationBroker extends AbstractAuthenticationBroker {
             };
         }
 
-            context.setSecurityContext(s);
+        return securityContext;
-            securityContexts.add(s);
-        }
-
-        try {
-            super.addConnection(context, info);
-        } catch (Exception e) {
-            securityContexts.remove(s);
-            context.setSecurityContext(null);
-            throw e;
-        }
     }
 }