QPID-6401 : [Java Broker] Add ability to validate connection attempts based on plugins

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

8 files changed, 149 insertions, 13 deletions

diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java
index 57b1d84a26..227089d722 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java
@@ -21,6 +21,7 @@ package org.apache.qpid.server.virtualhost.berkeleydb;
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
@@ -42,6 +43,7 @@ import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
 import org.apache.qpid.server.model.State;
 import org.apache.qpid.server.model.VirtualHostAlias;
 import org.apache.qpid.server.model.VirtualHostNode;
+import org.apache.qpid.server.protocol.AMQConnectionModel;
 import org.apache.qpid.server.protocol.LinkRegistry;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.stats.StatisticsCounter;
@@ -81,6 +83,13 @@ public class BDBHAReplicaVirtualHostImpl extends AbstractConfiguredObject<BDBHAR
     @ManagedAttributeField
     private int _housekeepingThreadCount;
 
+    @ManagedAttributeField
+    private List<String> _enabledConnectionValidators;
+
+    @ManagedAttributeField
+    private List<String> _disabledConnectionValidators;
+
+
     @ManagedObjectFactoryConstructor
     public BDBHAReplicaVirtualHostImpl(final Map<String, Object> attributes, VirtualHostNode<?> virtualHostNode)
     {
@@ -448,6 +457,24 @@ public class BDBHAReplicaVirtualHostImpl extends AbstractConfiguredObject<BDBHAR
     {
     }
 
+    @Override
+    public boolean authoriseCreateConnection(final AMQConnectionModel<?, ?> connection)
+    {
+        return false;
+    }
+
+    @Override
+    public List<String> getEnabledConnectionValidators()
+    {
+        return _enabledConnectionValidators;
+    }
+
+    @Override
+    public List<String> getDisabledConnectionValidators()
+    {
+        return _disabledConnectionValidators;
+    }
+
     private void throwUnsupportedForReplica()
     {
         throw new IllegalStateException("The virtual host state of " + getState()
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
index 79f37b66cb..aeb44ff671 100644
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
@@ -22,6 +22,7 @@ package org.apache.qpid.server.model;
 
 import java.security.AccessControlException;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 
@@ -42,6 +43,8 @@ public interface VirtualHost<X extends VirtualHost<X, Q, E>, Q extends Queue<?>,
     String STORE_TRANSACTION_OPEN_TIMEOUT_WARN  = "storeTransactionOpenTimeoutWarn";
     String HOUSE_KEEPING_THREAD_COUNT           = "houseKeepingThreadCount";
     String MODEL_VERSION                        = "modelVersion";
+    String ENABLED_CONNECTION_VALIDATORS        = "enabledConnectionValidators";
+    String DISABLED_CONNECTION_VALIDATORS       = "disabledConnectionValidators";
 
     @ManagedContextDefault( name = "queue.deadLetterQueueEnabled")
     public static final boolean DEFAULT_DEAD_LETTER_QUEUE_ENABLED = false;
@@ -88,6 +91,18 @@ public interface VirtualHost<X extends VirtualHost<X, Q, E>, Q extends Queue<?>,
     @DerivedAttribute( persist = true )
     String getModelVersion();
 
+    @ManagedContextDefault( name = "virtualhost.enabledConnectionValidators")
+    String DEFAULT_ENABLED_VALIDATORS = "[]";
+
+    @ManagedAttribute( defaultValue = "${virtualhost.enabledConnectionValidators}")
+    List<String> getEnabledConnectionValidators();
+
+    @ManagedContextDefault( name = "virtualhost.disabledConnectionValidators")
+    String DEFAULT_DISABLED_VALIDATORS = "[]";
+
+    @ManagedAttribute( defaultValue = "${virtualhost.disabledConnectionValidators}")
+    List<String> getDisabledConnectionValidators();
+
     @ManagedStatistic
     long getQueueCount();
 
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/ConnectionValidator.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/ConnectionValidator.java
new file mode 100644
index 0000000000..11f8944863
--- /dev/null
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/ConnectionValidator.java
@@ -0,0 +1,28 @@
+/*
+ *
+ * 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.qpid.server.plugin;
+
+import org.apache.qpid.server.protocol.AMQConnectionModel;
+
+public interface ConnectionValidator extends Pluggable
+{
+    boolean validateConnectionCreation(AMQConnectionModel<?, ?> connection);
+}
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
index 4086a67aae..21f0f47835 100644
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
@@ -63,6 +63,7 @@ import org.apache.qpid.server.message.MessageSource;
 import org.apache.qpid.server.message.ServerMessage;
 import org.apache.qpid.server.model.*;
 import org.apache.qpid.server.model.adapter.ConnectionAdapter;
+import org.apache.qpid.server.plugin.ConnectionValidator;
 import org.apache.qpid.server.plugin.QpidServiceLoader;
 import org.apache.qpid.server.plugin.SystemNodeCreator;
 import org.apache.qpid.server.protocol.AMQConnectionModel;
@@ -94,6 +95,8 @@ import org.apache.qpid.server.util.MapValueConverter;
 public abstract class AbstractVirtualHost<X extends AbstractVirtualHost<X>> extends AbstractConfiguredObject<X>
         implements VirtualHostImpl<X, AMQQueue<?>, ExchangeImpl<?>>, IConnectionRegistry.RegistryChangeListener, EventListener
 {
+    private final Collection<ConnectionValidator> _connectionValidators = new ArrayList<>();
+
     private static enum BlockingType { STORE, FILESYSTEM };
 
     private static final String USE_ASYNC_RECOVERY = "use_async_message_store_recovery";
@@ -162,6 +165,12 @@ public abstract class AbstractVirtualHost<X extends AbstractVirtualHost<X>> exte
     @ManagedAttributeField
     private int _housekeepingThreadCount;
 
+    @ManagedAttributeField
+    private List<String> _enabledConnectionValidators;
+
+    @ManagedAttributeField
+    private List<String> _disabledConnectionValidators;
+
 
     private boolean _useAsyncRecoverer;
 
@@ -297,6 +306,19 @@ public abstract class AbstractVirtualHost<X extends AbstractVirtualHost<X>> exte
 
 
         _fileSystemMaxUsagePercent = getContextValue(Integer.class, Broker.STORE_FILESYSTEM_MAX_USAGE_PERCENT);
+
+
+        QpidServiceLoader serviceLoader = new QpidServiceLoader();
+        for(ConnectionValidator validator : serviceLoader.instancesOf(ConnectionValidator.class))
+        {
+            if((_enabledConnectionValidators.isEmpty()
+                && (_disabledConnectionValidators.isEmpty()) || !_disabledConnectionValidators.contains(validator.getType()))
+               || _enabledConnectionValidators.contains(validator.getType()))
+            {
+                _connectionValidators.add(validator);
+            }
+
+        }
     }
 
     private void checkVHostStateIsActive()
@@ -438,6 +460,20 @@ public abstract class AbstractVirtualHost<X extends AbstractVirtualHost<X>> exte
         return _eventLogger;
     }
 
+    @Override
+    public boolean authoriseCreateConnection(final AMQConnectionModel<?, ?> connection)
+    {
+        getSecurityManager().authoriseCreateConnection(connection);
+        for(ConnectionValidator validator : _connectionValidators)
+        {
+            if(!validator.validateConnectionCreation(connection))
+            {
+                return false;
+            }
+        }
+        return true;
+    }
+
     /**
      * Initialise a housekeeping task to iterate over queues cleaning expired messages with no consumers
      * and checking for idle or open transactions that have exceeded the permitted thresholds.
@@ -526,6 +562,19 @@ public abstract class AbstractVirtualHost<X extends AbstractVirtualHost<X>> exte
     }
 
     @Override
+    public List<String> getEnabledConnectionValidators()
+    {
+        return _enabledConnectionValidators;
+    }
+
+    @Override
+    public List<String> getDisabledConnectionValidators()
+    {
+        return _disabledConnectionValidators;
+    }
+
+
+    @Override
     public AMQQueue<?> getQueue(String name)
     {
         return (AMQQueue<?>) getChildByName(Queue.class, name);
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java
index 29729b6c7d..b7689191bb 100755
--- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java
+++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java
@@ -33,6 +33,7 @@ import org.apache.qpid.server.message.MessageDestination;
 import org.apache.qpid.server.message.MessageSource;
 import org.apache.qpid.server.model.NoFactoryForTypeException;
 import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.protocol.AMQConnectionModel;
 import org.apache.qpid.server.protocol.LinkRegistry;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.security.SecurityManager;
@@ -108,4 +109,5 @@ public interface VirtualHostImpl< X extends VirtualHostImpl<X,Q,E>, Q extends AM
 
     EventLogger getEventLogger();
 
+    boolean authoriseCreateConnection(AMQConnectionModel<?, ?> connection);
 }
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
index 78228c209f..26d659fc34 100644
--- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
@@ -200,7 +200,12 @@ public class ServerConnectionDelegate extends ServerDelegate
             sconn.setVirtualHost(vhost);
             try
             {
-                vhost.getSecurityManager().authoriseCreateConnection(sconn);
+                if(!vhost.authoriseCreateConnection(sconn))
+                {
+                    sconn.setState(Connection.State.CLOSING);
+                    sconn.invoke(new ConnectionClose(ConnectionCloseCode.CONNECTION_FORCED, "Connection not authorized"));
+                    return;
+                }
             }
             catch (AccessControlException e)
             {
@@ -215,7 +220,8 @@ public class ServerConnectionDelegate extends ServerDelegate
         else
         {
             sconn.setState(Connection.State.CLOSING);
-            sconn.invoke(new ConnectionClose(ConnectionCloseCode.INVALID_PATH, "Unknown virtualhost '"+vhostName+"'"));
+            sconn.invoke(new ConnectionClose(ConnectionCloseCode.INVALID_PATH,
+                                             "Unknown virtualhost '" + vhostName + "'"));
         }
 
     }
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 3783cd70ac..92b6db9823 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
@@ -1564,23 +1564,30 @@ public class AMQProtocolEngine implements ServerProtocolEngine,
             else
             {
                 setVirtualHost(virtualHost);
-                try
+
+                if(virtualHost.authoriseCreateConnection(this))
                 {
-                    virtualHost.getSecurityManager().authoriseCreateConnection(this);
-                    if (getContextKey() == null)
+                    try
                     {
-                        setContextKey(new AMQShortString(Long.toString(System.currentTimeMillis())));
-                    }
+                        if (getContextKey() == null)
+                        {
+                            setContextKey(new AMQShortString(Long.toString(System.currentTimeMillis())));
+                        }
 
-                    MethodRegistry methodRegistry = getMethodRegistry();
-                    AMQMethodBody responseBody = methodRegistry.createConnectionOpenOkBody(virtualHostName);
+                        MethodRegistry methodRegistry = getMethodRegistry();
+                        AMQMethodBody responseBody = methodRegistry.createConnectionOpenOkBody(virtualHostName);
 
-                    writeFrame(responseBody.generateFrame(0));
-                    _state = ConnectionState.OPEN;
+                        writeFrame(responseBody.generateFrame(0));
+                        _state = ConnectionState.OPEN;
+                    }
+                    catch (AccessControlException e)
+                    {
+                        closeConnection(AMQConstant.ACCESS_REFUSED, e.getMessage(), 0);
+                    }
                 }
-                catch (AccessControlException e)
+                else
                 {
-                    closeConnection(AMQConstant.ACCESS_REFUSED, e.getMessage(),0);
+                    closeConnection(AMQConstant.ACCESS_REFUSED, "Connection refused",0);
                 }
             }
         }
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/Asserts.java b/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/Asserts.java
index 2674b248e1..de98d98e36 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/Asserts.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/systest/rest/Asserts.java
@@ -76,6 +76,8 @@ public class Asserts
                                 ConfiguredObject.DESCRIPTION,
                                 ConfiguredObject.CONTEXT,
                                 ConfiguredObject.DESIRED_STATE,
+                                VirtualHost.ENABLED_CONNECTION_VALIDATORS,
+                                VirtualHost.DISABLED_CONNECTION_VALIDATORS,
                                 VirtualHost.TYPE);
 
         assertEquals("Unexpected value of attribute " + VirtualHost.NAME,