QPID-5867: Add intruder protection functionality for a cluster of BDB HA virtual host nodes

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

2 files changed, 318 insertions, 0 deletions

diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java
index 80f6e7ea49..d830d488db 100644
--- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java
@@ -23,11 +23,13 @@ package org.apache.qpid.server.store.berkeleydb;
 import static org.mockito.Mockito.when;
 
 import java.io.File;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
@@ -38,8 +40,13 @@ import java.util.concurrent.atomic.AtomicReference;
 import com.sleepycat.je.rep.ReplicatedEnvironment;
 import com.sleepycat.je.rep.ReplicationConfig;
 
+import org.apache.qpid.server.configuration.IllegalConfigurationException;
 import org.apache.qpid.server.configuration.updater.TaskExecutor;
 import org.apache.qpid.server.configuration.updater.TaskExecutorImpl;
+import org.apache.qpid.server.logging.LogMessage;
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.logging.MessageLogger;
+import org.apache.qpid.server.logging.messages.VirtualHostMessages;
 import org.apache.qpid.server.model.Broker;
 import org.apache.qpid.server.model.BrokerModel;
 import org.apache.qpid.server.model.ConfigurationChangeListener;
@@ -505,6 +512,201 @@ public class BDBHAVirtualHostNodeTest extends QpidTestCase
 
     }
 
+    public void testIntruderProtection() throws Exception
+    {
+        int node1PortNumber = findFreePort();
+        String helperAddress = "localhost:" + node1PortNumber;
+        String groupName = "group";
+
+        Map<String, Object> node1Attributes = new HashMap<String, Object>();
+        node1Attributes.put(BDBHAVirtualHostNode.ID, UUID.randomUUID());
+        node1Attributes.put(BDBHAVirtualHostNode.TYPE, "BDB_HA");
+        node1Attributes.put(BDBHAVirtualHostNode.NAME, "node1");
+        node1Attributes.put(BDBHAVirtualHostNode.GROUP_NAME, groupName);
+        node1Attributes.put(BDBHAVirtualHostNode.ADDRESS, helperAddress);
+        node1Attributes.put(BDBHAVirtualHostNode.HELPER_ADDRESS, helperAddress);
+        node1Attributes.put(BDBHAVirtualHostNode.STORE_PATH, _bdbStorePath + File.separator + "1");
+
+        BDBHAVirtualHostNode<?> node1 = createAndStartHaVHN(node1Attributes);
+        BDBHAVirtualHost<?> host = (BDBHAVirtualHost<?>)node1.getVirtualHost();
+
+        List<String> permittedNodes = new ArrayList<String>();
+        int node2PortNumber = getNextAvailable(node1PortNumber+1);
+        permittedNodes.add(helperAddress);
+        permittedNodes.add("localhost:" + node2PortNumber);
+        host.setAttributes(Collections.<String, Object>singletonMap(BDBHAVirtualHost.PERMITTED_NODES, permittedNodes));
+
+        Map<String, Object> node2Attributes = new HashMap<String, Object>();
+        node2Attributes.put(BDBHAVirtualHostNode.ID, UUID.randomUUID());
+        node2Attributes.put(BDBHAVirtualHostNode.TYPE, "BDB_HA");
+        node2Attributes.put(BDBHAVirtualHostNode.NAME, "node2");
+        node2Attributes.put(BDBHAVirtualHostNode.GROUP_NAME, groupName);
+        node2Attributes.put(BDBHAVirtualHostNode.ADDRESS, "localhost:" + node2PortNumber);
+        node2Attributes.put(BDBHAVirtualHostNode.HELPER_ADDRESS, helperAddress);
+        node2Attributes.put(BDBHAVirtualHostNode.STORE_PATH, _bdbStorePath + File.separator + "2");
+        node2Attributes.put(BDBHAVirtualHostNode.HELPER_NODE_NAME, "node1");
+        node2Attributes.put(BDBHAVirtualHostNode.PRIORITY, 0);
+
+        BDBHAVirtualHostNode<?> node2 = createAndStartHaVHN(node2Attributes);
+
+        int node3PortNumber = getNextAvailable(node2PortNumber+1);
+        Map<String, Object> node3Attributes = new HashMap<String, Object>();
+        node3Attributes.put(BDBHAVirtualHostNode.ID, UUID.randomUUID());
+        node3Attributes.put(BDBHAVirtualHostNode.TYPE, "BDB_HA");
+        node3Attributes.put(BDBHAVirtualHostNode.NAME, "node3");
+        node3Attributes.put(BDBHAVirtualHostNode.GROUP_NAME, groupName);
+        node3Attributes.put(BDBHAVirtualHostNode.ADDRESS, "localhost:" + node3PortNumber);
+        node3Attributes.put(BDBHAVirtualHostNode.HELPER_ADDRESS, helperAddress);
+        node3Attributes.put(BDBHAVirtualHostNode.STORE_PATH, _bdbStorePath + File.separator + "3");
+        node3Attributes.put(BDBHAVirtualHostNode.HELPER_NODE_NAME, "node1");
+        node3Attributes.put(BDBHAVirtualHostNode.PRIORITY, 0);
+
+        try
+        {
+            createHaVHN(node3Attributes);
+            fail("The VHN should not be permitted to join the group");
+        }
+        catch(IllegalConfigurationException e)
+        {
+            assertEquals("Unexpected exception message", String.format("Node from '%s' is not permitted!", "localhost:" + node3PortNumber), e.getMessage());
+        }
+
+        // join node by skipping a step retrieving node state and checking the permitted hosts
+        node3Attributes.remove(BDBHAVirtualHostNode.HELPER_NODE_NAME);
+
+        final CountDownLatch stopLatch = new CountDownLatch(1);
+        ConfigurationChangeListener listener = new NoopConfigurationChangeListener()
+        {
+            @Override
+            public void stateChanged(ConfiguredObject<?> object, State oldState, State newState)
+            {
+                if (newState == State.ERRORED)
+                {
+                    stopLatch.countDown();
+                }
+            }
+        };
+        node1.addChangeListener(listener);
+
+        createHaVHN(node3Attributes);
+
+        assertTrue("Intruder protection was not triggered during expected timeout", stopLatch.await(10, TimeUnit.SECONDS));
+    }
+
+    public void testIntruderProtectionInManagementMode() throws Exception
+    {
+        int node1PortNumber = findFreePort();
+        String helperAddress = "localhost:" + node1PortNumber;
+        String groupName = "group";
+
+        Map<String, Object> node1Attributes = new HashMap<String, Object>();
+        node1Attributes.put(BDBHAVirtualHostNode.ID, UUID.randomUUID());
+        node1Attributes.put(BDBHAVirtualHostNode.TYPE, "BDB_HA");
+        node1Attributes.put(BDBHAVirtualHostNode.NAME, "node1");
+        node1Attributes.put(BDBHAVirtualHostNode.GROUP_NAME, groupName);
+        node1Attributes.put(BDBHAVirtualHostNode.ADDRESS, helperAddress);
+        node1Attributes.put(BDBHAVirtualHostNode.HELPER_ADDRESS, helperAddress);
+        node1Attributes.put(BDBHAVirtualHostNode.STORE_PATH, _bdbStorePath + File.separator + "1");
+
+        BDBHAVirtualHostNode<?> node1 = createAndStartHaVHN(node1Attributes);
+
+        int node2PortNumber = getNextAvailable(node1PortNumber+1);
+        Map<String, Object> node2Attributes = new HashMap<String, Object>();
+        node2Attributes.put(BDBHAVirtualHostNode.ID, UUID.randomUUID());
+        node2Attributes.put(BDBHAVirtualHostNode.TYPE, "BDB_HA");
+        node2Attributes.put(BDBHAVirtualHostNode.NAME, "node2");
+        node2Attributes.put(BDBHAVirtualHostNode.GROUP_NAME, groupName);
+        node2Attributes.put(BDBHAVirtualHostNode.ADDRESS, "localhost:" + node2PortNumber);
+        node2Attributes.put(BDBHAVirtualHostNode.HELPER_ADDRESS, helperAddress);
+        node2Attributes.put(BDBHAVirtualHostNode.STORE_PATH, _bdbStorePath + File.separator + "2");
+        node2Attributes.put(BDBHAVirtualHostNode.HELPER_NODE_NAME, "node1");
+        node2Attributes.put(BDBHAVirtualHostNode.PRIORITY, 0);
+
+        BDBHAVirtualHostNode<?> node2 = createAndStartHaVHN(node2Attributes);
+
+        final CountDownLatch stopLatch = new CountDownLatch(1);
+        ConfigurationChangeListener listener = new NoopConfigurationChangeListener()
+        {
+            @Override
+            public void stateChanged(ConfiguredObject<?> object, State oldState, State newState)
+            {
+                if (newState == State.ERRORED)
+                {
+                    stopLatch.countDown();
+                }
+            }
+        };
+        node1.addChangeListener(listener);
+
+        BDBHAVirtualHost<?> host = (BDBHAVirtualHost<?>)node1.getVirtualHost();
+
+        List<String> permittedNodes = new ArrayList<String>();
+        permittedNodes.add(helperAddress);
+        host.setAttributes(Collections.<String, Object>singletonMap(BDBHAVirtualHost.PERMITTED_NODES, permittedNodes));
+
+        assertTrue("Intruder protection was not triggered during expected timeout", stopLatch.await(10, TimeUnit.SECONDS));
+
+        when(_broker.isManagementMode()).thenReturn(true);
+        node1.start();
+
+        awaitRemoteNodes(node1, 1);
+
+        BDBHARemoteReplicationNode<?> remote = findRemoteNode(node1, node2.getName());
+        remote.delete();
+    }
+
+    public void testIntruderConnectedBeforePermittedNodesAreSet() throws Exception
+    {
+        int node1PortNumber = findFreePort();
+        String helperAddress = "localhost:" + node1PortNumber;
+        String groupName = "group";
+
+        Map<String, Object> node1Attributes = new HashMap<String, Object>();
+        node1Attributes.put(BDBHAVirtualHostNode.ID, UUID.randomUUID());
+        node1Attributes.put(BDBHAVirtualHostNode.TYPE, "BDB_HA");
+        node1Attributes.put(BDBHAVirtualHostNode.NAME, "node1");
+        node1Attributes.put(BDBHAVirtualHostNode.GROUP_NAME, groupName);
+        node1Attributes.put(BDBHAVirtualHostNode.ADDRESS, helperAddress);
+        node1Attributes.put(BDBHAVirtualHostNode.HELPER_ADDRESS, helperAddress);
+        node1Attributes.put(BDBHAVirtualHostNode.STORE_PATH, _bdbStorePath + File.separator + "1");
+
+        BDBHAVirtualHostNode<?> node1 = createAndStartHaVHN(node1Attributes);
+
+        int node2PortNumber = getNextAvailable(node1PortNumber+1);
+        Map<String, Object> node2Attributes = new HashMap<String, Object>();
+        node2Attributes.put(BDBHAVirtualHostNode.ID, UUID.randomUUID());
+        node2Attributes.put(BDBHAVirtualHostNode.TYPE, "BDB_HA");
+        node2Attributes.put(BDBHAVirtualHostNode.NAME, "node2");
+        node2Attributes.put(BDBHAVirtualHostNode.GROUP_NAME, groupName);
+        node2Attributes.put(BDBHAVirtualHostNode.ADDRESS, "localhost:" + node2PortNumber);
+        node2Attributes.put(BDBHAVirtualHostNode.HELPER_ADDRESS, helperAddress);
+        node2Attributes.put(BDBHAVirtualHostNode.STORE_PATH, _bdbStorePath + File.separator + "2");
+        node2Attributes.put(BDBHAVirtualHostNode.HELPER_NODE_NAME, "node1");
+
+        createAndStartHaVHN(node2Attributes);
+
+        final CountDownLatch stopLatch = new CountDownLatch(1);
+        ConfigurationChangeListener listener = new NoopConfigurationChangeListener()
+        {
+            @Override
+            public void stateChanged(ConfiguredObject<?> object, State oldState, State newState)
+            {
+                if (newState == State.ERRORED)
+                {
+                    stopLatch.countDown();
+                }
+            }
+        };
+        node1.addChangeListener(listener);
+
+        BDBHAVirtualHost<?> host = (BDBHAVirtualHost<?>)node1.getVirtualHost();
+        List<String> permittedNodes = new ArrayList<String>();
+        permittedNodes.add(helperAddress);
+        host.setAttributes(Collections.<String, Object>singletonMap(BDBHAVirtualHost.PERMITTED_NODES, permittedNodes));
+
+        assertTrue("Intruder protection was not triggered during expected timeout", stopLatch.await(20, TimeUnit.SECONDS));
+    }
+
     private BDBHARemoteReplicationNode<?> findRemoteNode(BDBHAVirtualHostNode<?> node, String name)
     {
         for (RemoteReplicationNode<?> remoteNode : node.getRemoteReplicationNodes())
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/replication/ReplicatedEnvironmentFacadeTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/replication/ReplicatedEnvironmentFacadeTest.java
index 86d47a6a8b..89e618482b 100644
--- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/replication/ReplicatedEnvironmentFacadeTest.java
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/replication/ReplicatedEnvironmentFacadeTest.java
@@ -24,14 +24,19 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import java.io.File;
+import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.apache.log4j.Logger;
+import org.apache.qpid.server.configuration.IllegalConfigurationException;
 import org.apache.qpid.server.store.berkeleydb.EnvironmentFacade;
 import org.apache.qpid.test.utils.QpidTestCase;
 import org.apache.qpid.test.utils.TestFileUtils;
@@ -41,6 +46,7 @@ import com.sleepycat.je.Database;
 import com.sleepycat.je.DatabaseConfig;
 import com.sleepycat.je.Durability;
 import com.sleepycat.je.Environment;
+import com.sleepycat.je.EnvironmentConfig;
 import com.sleepycat.je.Transaction;
 import com.sleepycat.je.rep.NodeState;
 import com.sleepycat.je.rep.ReplicatedEnvironment;
@@ -52,6 +58,7 @@ import com.sleepycat.je.rep.StateChangeListener;
 
 public class ReplicatedEnvironmentFacadeTest extends QpidTestCase
 {
+    private static final Logger LOGGER = Logger.getLogger(ReplicatedEnvironmentFacadeTest.class);
     private static final int TEST_NODE_PORT = new QpidTestCase().findFreePort();
     private static final int LISTENER_TIMEOUT = 5;
     private static final int WAIT_STATE_CHANGE_TIMEOUT = 30;
@@ -644,6 +651,105 @@ public class ReplicatedEnvironmentFacadeTest extends QpidTestCase
         }
     }
 
+    public void testSetPermittedNodes() throws Exception
+    {
+        ReplicatedEnvironmentFacade firstNode = createMaster();
+
+        Set<String> permittedNodes = new HashSet<String>();
+        permittedNodes.add("localhost:" + TEST_NODE_PORT);
+        permittedNodes.add("localhost:" + getNextAvailable(TEST_NODE_PORT + 1));
+        firstNode.setPermittedNodes(permittedNodes);
+
+        NodeState nodeState = firstNode.getRemoteNodeState(new ReplicatedEnvironmentFacade.ReplicationNodeImpl(TEST_NODE_NAME, TEST_NODE_HOST_PORT));
+
+        Collection<String> appStatePermittedNodes = firstNode.bytesToPermittedNodeList(nodeState.getAppState());
+        assertEquals("Unexpected permitted nodes", permittedNodes, new HashSet<String>(appStatePermittedNodes));
+    }
+
+    public void testPermittedNodeIsAllowedToConnect() throws Exception
+    {
+        ReplicatedEnvironmentFacade firstNode = createMaster();
+
+        int replica1Port = getNextAvailable(TEST_NODE_PORT + 1);
+        String node1NodeHostPort = "localhost:" + replica1Port;
+
+        Set<String> permittedNodes = new HashSet<String>();
+        permittedNodes.add("localhost:" + TEST_NODE_PORT);
+        permittedNodes.add(node1NodeHostPort);
+        firstNode.setPermittedNodes(permittedNodes);
+
+        ReplicatedEnvironmentConfiguration configuration =  createReplicatedEnvironmentConfiguration(TEST_NODE_NAME + "_1", node1NodeHostPort, false);
+        when(configuration.getHelperNodeName()).thenReturn(TEST_NODE_NAME);
+
+        TestStateChangeListener stateChangeListener = new TestStateChangeListener(State.REPLICA);
+        ReplicatedEnvironmentFacade secondNode = createReplicatedEnvironmentFacade(TEST_NODE_NAME + "_1",
+                stateChangeListener, new NoopReplicationGroupListener(), configuration);
+        assertTrue("Environment was not created", stateChangeListener.awaitForStateChange(LISTENER_TIMEOUT, TimeUnit.SECONDS));
+        assertEquals("Unexpected state", State.REPLICA.name(), secondNode.getNodeState());
+    }
+
+    public void testNotPermittedNodeIsNotAllowedToConnect() throws Exception
+    {
+        ReplicatedEnvironmentFacade firstNode = createMaster();
+
+        int replica1Port = getNextAvailable(TEST_NODE_PORT + 1);
+        String node1NodeHostPort = "localhost:" + replica1Port;
+
+        Set<String> permittedNodes = new HashSet<String>();
+        permittedNodes.add("localhost:" + TEST_NODE_PORT);
+
+        firstNode.setPermittedNodes(permittedNodes);
+
+        ReplicatedEnvironmentConfiguration configuration =  createReplicatedEnvironmentConfiguration(TEST_NODE_NAME + "_1", node1NodeHostPort, false);
+        when(configuration.getHelperNodeName()).thenReturn(TEST_NODE_NAME);
+
+        try
+        {
+            createReplicatedEnvironmentFacade(TEST_NODE_NAME + "_1", new TestStateChangeListener(State.REPLICA), new NoopReplicationGroupListener(), configuration);
+            fail("Node is not allowed to connect from " + node1NodeHostPort + " but environment was successfully created");
+        }
+        catch (IllegalConfigurationException e)
+        {
+            assertEquals("Unexpected exception message", String.format("Node from '%s' is not permitted!",
+                    node1NodeHostPort), e.getMessage());
+        }
+    }
+
+    public void testIntruderNodeIsDetected() throws Exception
+    {
+        final CountDownLatch intruderLatch = new CountDownLatch(1);
+        ReplicationGroupListener listener = new NoopReplicationGroupListener()
+        {
+            @Override
+            public void onIntruderNode(ReplicationNode node)
+            {
+                intruderLatch.countDown();
+            }
+        };
+        ReplicatedEnvironmentFacade firstNode = createMaster(listener);
+        int replica1Port = getNextAvailable(TEST_NODE_PORT + 1);
+        String node1NodeHostPort = "localhost:" + replica1Port;
+
+        Set<String> permittedNodes = new HashSet<String>();
+        permittedNodes.add("localhost:" + TEST_NODE_PORT);
+
+        firstNode.setPermittedNodes(permittedNodes);
+
+        String nodeName = TEST_NODE_NAME + "_1";
+        File environmentPathFile = new File(_storePath, nodeName);
+        environmentPathFile.mkdirs();
+
+        ReplicationConfig replicationConfig = new ReplicationConfig(TEST_GROUP_NAME, TEST_NODE_NAME + "_1", node1NodeHostPort);
+        replicationConfig.setHelperHosts(TEST_NODE_HOST_PORT);
+
+        EnvironmentConfig envConfig = new EnvironmentConfig();
+        envConfig.setAllowCreate(true);
+        envConfig.setTransactional(true);
+        envConfig.setDurability(TEST_DURABILITY);
+        ReplicatedEnvironment intruder = new ReplicatedEnvironment(environmentPathFile, replicationConfig, envConfig);
+        assertTrue("Intruder node was not detected", intruderLatch.await(10, TimeUnit.SECONDS));
+    }
+
     private ReplicatedEnvironmentFacade createMaster() throws Exception
     {
         return createMaster(new NoopReplicationGroupListener());
@@ -678,6 +784,10 @@ public class ReplicatedEnvironmentFacadeTest extends QpidTestCase
                                                 StateChangeListener stateChangeListener, ReplicationGroupListener replicationGroupListener)
     {
         ReplicatedEnvironmentConfiguration config = createReplicatedEnvironmentConfiguration(nodeName, nodeHostPort, designatedPrimary);
+        return createReplicatedEnvironmentFacade(nodeName, stateChangeListener, replicationGroupListener, config);
+    }
+
+    private ReplicatedEnvironmentFacade createReplicatedEnvironmentFacade(String nodeName, StateChangeListener stateChangeListener, ReplicationGroupListener replicationGroupListener, ReplicatedEnvironmentConfiguration config) {
         ReplicatedEnvironmentFacade ref = new ReplicatedEnvironmentFacade(config);
         ref.setStateChangeListener(stateChangeListener);
         ref.setReplicationGroupListener(replicationGroupListener);
@@ -735,5 +845,11 @@ public class ReplicatedEnvironmentFacadeTest extends QpidTestCase
         {
         }
 
+        @Override
+        public void onIntruderNode(ReplicationNode node)
+        {
+            LOGGER.warn("Intruder node " + node);
+        }
+
     }
 }