Report authentication failures as returned by the broker

Author: Emile Joubert

src/com/rabbitmq/client/AuthenticationFailureException.java

@@ -0,0 +1,28 @@
+//  The contents of this file are subject to the Mozilla Public License
+//  Version 1.1 (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.mozilla.org/MPL/
+//
+//  Software distributed under the License is distributed on an "AS IS"
+//  basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+//  the License for the specific language governing rights and
+//  limitations under the License.
+//
+//  The Original Code is RabbitMQ.
+//
+//  The Initial Developer of the Original Code is GoPivotal, Inc.
+//  Copyright (c) 2013-2013 GoPivotal, Inc.  All rights reserved.
+//
+
+package com.rabbitmq.client;
+
+/**
+ * Thrown when the broker refuses access due to an authentication failure.
+ */
+
+public class AuthenticationFailureException extends Exception
+{
+    public AuthenticationFailureException(String reason) {
+        super(reason);
+    }
+}

src/com/rabbitmq/client/ConnectionFactory.java

@@ -484,7 +484,7 @@ protected void configureSocket(Socket socket) throws IOException{
      * @return an interface to the connection
      * @throws IOException if it encounters a problem
      */
-    public Connection newConnection(Address[] addrs) throws IOException {
+    public Connection newConnection(Address[] addrs) throws IOException, AuthenticationFailureException {
         return newConnection(null, addrs);
     }
 
@@ -494,10 +494,10 @@ public Connection newConnection(Address[] addrs) throws IOException {
      * @param addrs an array of known broker addresses (hostname/port pairs) to try in order
      * @return an interface to the connection
      * @throws IOException if it encounters a problem
+     * @throws AuthenticationFailureException if the login fails
      */
     public Connection newConnection(ExecutorService executor, Address[] addrs)
-        throws IOException
-    {
+            throws IOException, AuthenticationFailureException {
         IOException lastException = null;
         for (Address addr : addrs) {
             try {
@@ -528,8 +528,9 @@ public Connection newConnection(ExecutorService executor, Address[] addrs)
      * Create a new broker connection
      * @return an interface to the connection
      * @throws IOException if it encounters a problem
+     * @throws AuthenticationFailureException if the login fails
      */
-    public Connection newConnection() throws IOException {
+    public Connection newConnection() throws IOException, AuthenticationFailureException {
         return newConnection(null,
                              new Address[] {new Address(getHost(), getPort())}
                             );
@@ -540,8 +541,9 @@ public Connection newConnection() throws IOException {
      * @param executor thread execution service for consumers on the connection
      * @return an interface to the connection
      * @throws IOException if it encounters a problem
+     * @throws AuthenticationFailureException if the login fails
      */
-    public Connection newConnection(ExecutorService executor) throws IOException {
+    public Connection newConnection(ExecutorService executor) throws IOException, AuthenticationFailureException {
         return newConnection(executor,
                              new Address[] {new Address(getHost(), getPort())}
                             );

src/com/rabbitmq/client/PossibleAuthenticationFailureException.java

@@ -19,7 +19,8 @@
 import java.io.IOException;
 
 /**
- * Thrown when the likely cause is an authentication failure.
+ * Thrown when the likely cause is an authentication failure and the broker
+ * communicates this by closing the connection.
  */
 public class PossibleAuthenticationFailureException extends IOException
 {

src/com/rabbitmq/client/impl/AMQConnection.java

@@ -30,6 +30,7 @@
 import java.util.concurrent.TimeoutException;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.AuthenticationFailureException;
 import com.rabbitmq.client.BlockedListener;
 import com.rabbitmq.client.Method;
 import com.rabbitmq.client.AlreadyClosedException;
@@ -85,6 +86,7 @@ public static final Map<String, Object> defaultClientProperties() {
         capabilities.put("basic.nack", true);
         capabilities.put("consumer_cancel_notify", true);
         capabilities.put("connection.blocked", true);
+        capabilities.put("authentication_failure_close", true);
 
         props.put("capabilities", capabilities);
 
@@ -277,15 +279,16 @@ public AMQConnection(String username,
      * and frame max values after tuning has taken place.
      * @throws IOException if an error is encountered
      * either before, or during, protocol negotiation;
+     * @throws AuthenticationFailureException if the broker closes the
+     * connection with ACCESS_REFUSED.
      * sub-classes {@link ProtocolVersionMismatchException} and
      * {@link PossibleAuthenticationFailureException} will be thrown in the
      * corresponding circumstances. If an exception is thrown, connection
      * resources allocated can all be garbage collected when the connection
      * object is no longer referenced.
      */
     public void start()
-        throws IOException
-    {
+            throws IOException, AuthenticationFailureException {
         this._running = true;
         // Make sure that the first thing we do is to send the header,
         // which should cause any socket errors to show up for us, rather
@@ -357,6 +360,16 @@ public void start()
                         response = sm.handleChallenge(challenge, this.username, this.password);
                     }
                 } catch (ShutdownSignalException e) {
+                    Object shutdownReason = e.getReason();
+                    if (shutdownReason instanceof AMQCommand) {
+                        Method shutdownMethod = ((AMQCommand) shutdownReason).getMethod();
+                        if (shutdownMethod instanceof AMQP.Connection.Close) {
+                            AMQP.Connection.Close shutdownClose =  (AMQP.Connection.Close) shutdownMethod;
+                            if (shutdownClose.getReplyCode() == AMQP.ACCESS_REFUSED) {
+                                throw new AuthenticationFailureException(shutdownClose.getReplyText());
+                            }
+                        }
+                    }
                     throw new PossibleAuthenticationFailureException(e);
                 }
             } while (connTune == null);
@@ -645,7 +658,7 @@ public boolean processControlCommand(Command c) throws IOException
     }
 
     public void handleConnectionClose(Command closeCommand) {
-        ShutdownSignalException sse = shutdown(closeCommand, false, null, false);
+        ShutdownSignalException sse = shutdown(closeCommand, false, null, _inConnectionNegotiation);
         try {
             _channel0.quiescingTransmit(new AMQP.Connection.CloseOk.Builder().build());
         } catch (IOException _) { } // ignore

test/src/com/rabbitmq/client/test/AMQConnectionTest.java

@@ -28,6 +28,7 @@
 import junit.framework.TestCase;
 import junit.framework.TestSuite;
 
+import com.rabbitmq.client.AuthenticationFailureException;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
@@ -81,7 +82,7 @@ public static TestSuite suite() {
     /** Check the AMQConnection does send exactly 1 initial header, and deal correctly with
      * the frame handler throwing an exception when we try to read data
      */
-    public void testConnectionSendsSingleHeaderAndTimesOut() {
+    public void testConnectionSendsSingleHeaderAndTimesOut() throws AuthenticationFailureException {
         IOException exception = new SocketTimeoutException();
         _mockFrameHandler.setExceptionOnReadingFrames(exception);
         MyExceptionHandler handler = new MyExceptionHandler();
@@ -124,7 +125,7 @@ public void testConnectionSendsSingleHeaderAndTimesOut() {
     /**
      * Test that we catch timeout between connect and negotiation of the connection being finished.
      */
-    public void testConnectionHangInNegotiation() {
+    public void testConnectionHangInNegotiation() throws AuthenticationFailureException {
         this._mockFrameHandler.setTimeoutCount(10); // to limit hang
         MyExceptionHandler handler = new MyExceptionHandler();
         assertEquals(0, this._mockFrameHandler.countHeadersSent());

test/src/com/rabbitmq/client/test/BrokerTestCase.java

@@ -30,6 +30,7 @@
 import com.rabbitmq.client.Method;
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.client.AlreadyClosedException;
+import com.rabbitmq.client.AuthenticationFailureException;
 import com.rabbitmq.client.impl.ShutdownNotifierComponent;
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.tools.Host;
@@ -81,7 +82,7 @@ protected void releaseResources()
     }
 
     protected void restart()
-            throws IOException {
+            throws IOException, AuthenticationFailureException {
         tearDown();
         bareRestart();
         setUp();
@@ -95,7 +96,11 @@ protected void bareRestart()
     public void openConnection()
             throws IOException {
         if (connection == null) {
-            connection = connectionFactory.newConnection();
+            try {
+                connection = connectionFactory.newConnection();
+            } catch (AuthenticationFailureException afe) {
+                fail("Unexpected authentication failure");
+            }
         }
     }
 

test/src/com/rabbitmq/client/test/functional/ClusteredTestBase.java

@@ -17,6 +17,7 @@
 package com.rabbitmq.client.test.functional;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.AuthenticationFailureException;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
@@ -63,6 +64,9 @@ public void openConnection() throws IOException {
             catch (IOException e) {
                 // Must be no secondary node
             }
+            catch (AuthenticationFailureException afe)  {
+                fail("Unexpected authentication failure");
+            }
         }
 
         if (clusteredConnection != null &&

test/src/com/rabbitmq/client/test/functional/ConnectionOpen.java

@@ -22,6 +22,7 @@
 import java.net.Socket;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.AuthenticationFailureException;
 import com.rabbitmq.client.MalformedFrameException;
 import com.rabbitmq.client.Method;
 import com.rabbitmq.client.impl.SocketFrameHandler;
@@ -85,7 +86,7 @@ public void testCrazyProtocolHeader() throws IOException {
         }
     }
 
-    public void testFrameMaxLessThanFrameMinSize() throws IOException {
+    public void testFrameMaxLessThanFrameMinSize() throws IOException, AuthenticationFailureException {
         ConnectionFactory factory = new ConnectionFactory();
         factory.setRequestedFrameMax(100);
         try {

test/src/com/rabbitmq/client/test/functional/FrameMax.java

@@ -24,6 +24,7 @@
 import java.util.concurrent.ExecutorService;
 
 import com.rabbitmq.client.Address;
+import com.rabbitmq.client.AuthenticationFailureException;
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
@@ -71,7 +72,7 @@ public void testFrameSizes()
     /* server should reject frames larger than AMQP.FRAME_MIN_SIZE
      * during connection negotiation */
     public void testRejectLargeFramesDuringConnectionNegotiation()
-        throws IOException
+        throws IOException, AuthenticationFailureException
     {
         ConnectionFactory cf = new ConnectionFactory();
         cf.getClientProperties().put("too_long", LongStringHelper.asLongString(new byte[AMQP.FRAME_MIN_SIZE]));
@@ -85,7 +86,7 @@ public void testRejectLargeFramesDuringConnectionNegotiation()
     /* server should reject frames larger than the negotiated frame
      * size */
     public void testRejectExceedingFrameMax()
-        throws IOException
+        throws IOException, AuthenticationFailureException
     {
         closeChannel();
         closeConnection();
@@ -161,7 +162,7 @@ public GenerousAMQConnection(ConnectionFactory factory,
     private static class GenerousConnectionFactory extends ConnectionFactory {
 
         @Override public Connection newConnection(ExecutorService executor, Address[] addrs)
-            throws IOException
+            throws IOException, AuthenticationFailureException
         {
             IOException lastException = null;
             for (Address addr : addrs) {

test/src/com/rabbitmq/client/test/functional/QueueExclusivity.java

@@ -21,6 +21,7 @@
 import java.util.HashMap;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.AuthenticationFailureException;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.QueueingConsumer;
@@ -36,7 +37,11 @@ public class QueueExclusivity extends BrokerTestCase {
     String q = "exclusiveQ";
 
     protected void createResources() throws IOException {
-        altConnection = connectionFactory.newConnection();
+        try {
+            altConnection = connectionFactory.newConnection();
+        } catch (AuthenticationFailureException afe) {
+            fail("Unexpected authentication failure");
+        }
         altChannel = altConnection.createChannel();
         altChannel.queueDeclare(q,
         // not durable, exclusive, not auto-delete