QPID-5745 : [Java Broker] enforce disconnect of AMQP 0-x connections when authentication is not performed in a timely manner

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1616899 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 54 insertions, 15 deletions

diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java
index cefd1ee0b2..8ce014dacc 100755
--- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java
@@ -20,8 +20,16 @@
  */
 package org.apache.qpid.server.protocol.v0_10;
 
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+
+import javax.security.auth.Subject;
+
+import org.apache.log4j.Logger;
+
 import org.apache.qpid.protocol.ServerProtocolEngine;
-import org.apache.qpid.server.logging.EventLogger;
 import org.apache.qpid.server.logging.messages.ConnectionMessages;
 import org.apache.qpid.server.model.Port;
 import org.apache.qpid.server.model.Transport;
@@ -31,16 +39,12 @@ import org.apache.qpid.transport.network.Disassembler;
 import org.apache.qpid.transport.network.InputHandler;
 import org.apache.qpid.transport.network.NetworkConnection;
 
-import javax.security.auth.Subject;
-import java.net.SocketAddress;
-import java.nio.ByteBuffer;
-import java.security.AccessController;
-import java.security.PrivilegedAction;
-
 
 public class ProtocolEngine_0_10  extends InputHandler implements ServerProtocolEngine
 {
     public static final int MAX_FRAME_SIZE = 64 * 1024 - 1;
+    private static final Logger _logger = Logger.getLogger(ProtocolEngine_0_10.class);
+
 
     private NetworkConnection _network;
     private long _readBytes;
@@ -154,6 +158,26 @@ public class ProtocolEngine_0_10  extends InputHandler implements ServerProtocol
     public void received(final ByteBuffer buf)
     {
         _lastReadTime = System.currentTimeMillis();
+        if(_connection.getAuthorizedPrincipal() == null &&
+           (_lastReadTime - _createTime) > _connection.getPort().getContextValue(Long.class,
+                                                                                 Port.CONNECTION_MAXIMUM_AUTHENTICATION_DELAY) )
+        {
+            Subject.doAs(_connection.getAuthorizedSubject(), new PrivilegedAction<Object>()
+            {
+                @Override
+                public Object run()
+                {
+
+                    _logger.warn("Connection has taken more than "
+                                 + _connection.getPort()
+                            .getContextValue(Long.class, Port.CONNECTION_MAXIMUM_AUTHENTICATION_DELAY)
+                                 + "ms to establish identity.  Closing as possible DoS.");
+                    _connection.getEventLogger().message(ConnectionMessages.IDLE_CLOSE());
+                    _network.close();
+                    return null;
+                }
+            });
+        }
         super.received(buf);
         _connection.receivedComplete();
     }
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
index 2ad79ad980..8ddd04f51a 100644
--- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
@@ -75,7 +75,7 @@ public class ServerConnection extends Connection implements AMQConnectionModel<S
     private final long _connectionId;
     private final Object _reference = new Object();
     private VirtualHostImpl _virtualHost;
-    private Port _port;
+    private Port<?> _port;
     private AtomicLong _lastIoTime = new AtomicLong();
     private boolean _blocking;
     private Transport _transport;
@@ -189,12 +189,12 @@ public class ServerConnection extends Connection implements AMQConnectionModel<S
     }
 
     @Override
-    public Port getPort()
+    public Port<?> getPort()
     {
         return _port;
     }
 
-    public void setPort(Port port)
+    public void setPort(Port<?> port)
     {
         _port = port;
     }
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
index 3ae9de8368..484ca6f404 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
@@ -93,7 +93,8 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
     // channels.  This value must be of the form 2^x - 1.
     private static final int CHANNEL_CACHE_SIZE = 0xff;
     private static final int REUSABLE_BYTE_BUFFER_CAPACITY = 65 * 1024;
-    private final Port _port;
+    private final Port<?> _port;
+    private final long _creationTime;
 
     private AMQShortString _contextKey;
 
@@ -166,12 +167,13 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
 
     private final ReentrantLock _receivedLock;
     private AtomicLong _lastWriteTime = new AtomicLong(System.currentTimeMillis());
-    private final Broker _broker;
+    private final Broker<?> _broker;
     private final Transport _transport;
 
     private volatile boolean _closeWhenNoRoute;
     private volatile boolean _stopped;
     private long _readBytes;
+    private boolean _authenticated;
 
     public AMQProtocolEngine(Broker broker,
                              final NetworkConnection network,
@@ -210,6 +212,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
         _dataDelivered = new StatisticsCounter("data-delivered-" + getSessionID());
         _messagesReceived = new StatisticsCounter("messages-received-" + getSessionID());
         _dataReceived = new StatisticsCounter("data-received-" + getSessionID());
+        _creationTime = System.currentTimeMillis();
     }
 
     private <T> T runAsSubject(PrivilegedAction<T> action)
@@ -277,7 +280,18 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
             @Override
             public Void run()
             {
+
                 final long arrivalTime = System.currentTimeMillis();
+                if(!_authenticated &&
+                   (arrivalTime - _creationTime) > _port.getContextValue(Long.class,
+                                                                         Port.CONNECTION_MAXIMUM_AUTHENTICATION_DELAY))
+                {
+                    _logger.warn("Connection has taken more than "
+                                 + _port.getContextValue(Long.class, Port.CONNECTION_MAXIMUM_AUTHENTICATION_DELAY)
+                                 + "ms to establish identity.  Closing as possible DoS.");
+                    getEventLogger().message(ConnectionMessages.IDLE_CLOSE());
+                    closeProtocolSession();
+                }
                 _lastReceivedTime = arrivalTime;
                 _lastIoTime = arrivalTime;
                 _readBytes += msg.remaining();
@@ -1200,6 +1214,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
             throw new IllegalArgumentException("authorizedSubject cannot be null");
         }
 
+        _authenticated = true;
         _authorizedSubject.getPrincipals().addAll(authorizedSubject.getPrincipals());
         _authorizedSubject.getPrivateCredentials().addAll(authorizedSubject.getPrivateCredentials());
         _authorizedSubject.getPublicCredentials().addAll(authorizedSubject.getPublicCredentials());
@@ -1353,7 +1368,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, AMQProtocolSessi
     }
 
     @Override
-    public Port getPort()
+    public Port<?> getPort()
     {
         return _port;
     }
diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
index ffa65b2477..2a48ccb2df 100644
--- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
+++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
@@ -62,7 +62,7 @@ import org.apache.qpid.server.virtualhost.VirtualHostImpl;
 public class Connection_1_0 implements ConnectionEventListener, AMQConnectionModel<Connection_1_0,Session_1_0>
 {
 
-    private final Port _port;
+    private final Port<?> _port;
     private final Broker _broker;
     private final SubjectCreator _subjectCreator;
     private VirtualHostImpl _vhost;
@@ -358,7 +358,7 @@ public class Connection_1_0 implements ConnectionEventListener, AMQConnectionMod
     }
 
     @Override
-    public Port getPort()
+    public Port<?> getPort()
     {
         return _port;
     }