Merged from trunk up to r826646

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/java-broker-0-10@829268 13f79535-47bb-0310-9956-ffa450edef68

10 files changed, 210 insertions, 24 deletions

diff --git a/qpid/java/broker/bin/qpid-server b/qpid/java/broker/bin/qpid-server
index 738fc6e084..4e11fd5fe7 100755
--- a/qpid/java/broker/bin/qpid-server
+++ b/qpid/java/broker/bin/qpid-server
@@ -26,11 +26,6 @@ fi
 # Set classpath to include Qpid jar with all required jars in manifest
 QPID_LIBS=$QPID_HOME/lib/qpid-all.jar:$QPID_HOME/lib/bdbstore-launch.jar
 
-# Default Log4j to append to its log file
-if [ -z "$QPID_LOG_APPEND" ]; then
-    export QPID_LOG_APPEND="true"
-fi
-
 # Set other variables used by the qpid-run script before calling
 export JAVA=java \
        JAVA_VM=-server \
diff --git a/qpid/java/broker/etc/log4j.xml b/qpid/java/broker/etc/log4j.xml
index 4b71772a0e..484b203df8 100644
--- a/qpid/java/broker/etc/log4j.xml
+++ b/qpid/java/broker/etc/log4j.xml
@@ -50,7 +50,7 @@
 
     <appender class="org.apache.log4j.FileAppender" name="FileAppender">
         <param name="File" value="${QPID_WORK}/log/${logprefix}qpid${logsuffix}.log"/>
-        <param name="Append" value="${QPID_LOG_APPEND}"/>
+        <param name="Append" value="false"/>
 
         <layout class="org.apache.log4j.PatternLayout">
             <param name="ConversionPattern" value="%d %-5p [%t] %C{2} (%F:%L) - %m%n"/>
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
index f9641d6eab..1d0ca19e91 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
@@ -679,6 +679,7 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol
     /** This must be called when the session is _closed in order to free up any resources managed by the session. */
     public void closeSession() throws AMQException
     {
+        // REMOVE THIS SHOULD NOT BE HERE. 
         if (CurrentActor.get() == null)
         {
             CurrentActor.set(_actor);
@@ -694,6 +695,8 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol
             if (_managedObject != null)
             {
                 _managedObject.unregister();
+                // Ensure we only do this once.
+                _managedObject = null;
             }
 
             for (Task task : _taskList)
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/PrincipalPermissions.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/PrincipalPermissions.java
index 42e662b85a..0b95d213f4 100755
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/PrincipalPermissions.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/PrincipalPermissions.java
@@ -53,6 +53,7 @@ public class PrincipalPermissions
     private static final int PUBLISH_EXCHANGES_KEY = 0;
 
     private Map _permissions;
+    private boolean _fullVHostAccess = false;
 
     private String _user;
 
@@ -82,6 +83,9 @@ public class PrincipalPermissions
     {
         switch (permission)
         {
+            case ACCESS:// Parameters : None
+                grantAccess(permission);
+                break;
             case CONSUME: // Parameters : AMQShortString queueName, Boolean Temporary, Boolean ownQueueOnly
                 grantConsume(permission, parameters);
                 break;
@@ -98,7 +102,6 @@ public class PrincipalPermissions
                 break;
             /* The other cases just fall through to no-op */
             case DELETE:
-            case ACCESS: // This is a no-op as the existence of this PrincipalPermission object is scoped per VHost for ACCESS
             case BIND: // All the details are currently included in the create setup.
             case PURGE:
             case UNBIND:
@@ -107,6 +110,11 @@ public class PrincipalPermissions
 
     }
 
+    private void grantAccess(Permission permission)
+    {
+        _fullVHostAccess = true;
+    }
+    
 	private void grantPublish(Permission permission, Object... parameters) {
 		Map publishRights = (Map) _permissions.get(permission);
 
@@ -353,9 +361,8 @@ public class PrincipalPermissions
 
         switch (permission)
         {
-            case ACCESS:
-                return AuthzResult.ALLOWED; // This is here for completeness but the SimpleXML ACLManager never calls it.
-                // The existence of this user specific PP can be validated in the map SimpleXML maintains.
+            case ACCESS://No Parameters 
+                return AuthzResult.ALLOWED; // The existence of this user-specific PP infers some level of access is authorised
             case BIND: // Parameters : QueueBindMethod , Exchange , AMQQueue, AMQShortString routingKey
                 return authoriseBind(parameters);
             case CREATEQUEUE:// Parameters : boolean autodelete, AMQShortString name
@@ -371,12 +378,28 @@ public class PrincipalPermissions
             case PURGE:
             case UNBIND:
             default:
-                return AuthzResult.DENIED;
+                if(_fullVHostAccess)
+                {
+                    //user has been granted full access to the vhost
+                    return AuthzResult.ALLOWED;
+                }
+                else
+                {
+                    //SimpleXML ACL does not implement these permissions and should abstain
+                    return AuthzResult.ABSTAIN;
+                }
         }
 
     }
 
-	private AuthzResult authoriseConsume(Permission permission, Object... parameters) {
+	private AuthzResult authoriseConsume(Permission permission, Object... parameters)
+	{
+	    if(_fullVHostAccess)
+	    {
+	        //user has been granted full access to the vhost
+	        return AuthzResult.ALLOWED;
+	    }
+	    
 		if (parameters.length == 1 && parameters[0] instanceof AMQQueue)
 		{
 		    AMQQueue queue = ((AMQQueue) parameters[0]);
@@ -435,8 +458,15 @@ public class PrincipalPermissions
 		return AuthzResult.DENIED;
 	}
 
-	private AuthzResult authorisePublish(Permission permission, Object... parameters) {
-		Map publishRights = (Map) _permissions.get(permission);
+	private AuthzResult authorisePublish(Permission permission, Object... parameters)
+	{
+	    if(_fullVHostAccess)
+	    {
+	        //user has been granted full access to the vhost
+	        return AuthzResult.ALLOWED;
+	    }
+
+	    Map publishRights = (Map) _permissions.get(permission);
 
 		if (publishRights == null)
 		{
@@ -495,7 +525,14 @@ public class PrincipalPermissions
 		}
 	}
 
-	private AuthzResult authoriseCreateExchange(Permission permission, Object... parameters) {
+	private AuthzResult authoriseCreateExchange(Permission permission, Object... parameters)
+	{
+        if(_fullVHostAccess)
+        {
+            //user has been granted full access to the vhost
+            return AuthzResult.ALLOWED;
+        }
+
 		Map rights = (Map) _permissions.get(permission);
 
 		AMQShortString exchangeName = (AMQShortString) parameters[0];
@@ -512,8 +549,15 @@ public class PrincipalPermissions
 		}
 	}
 
-	private AuthzResult authoriseCreateQueue(Permission permission, Object... parameters) {
-		Map createRights = (Map) _permissions.get(permission);
+	private AuthzResult authoriseCreateQueue(Permission permission, Object... parameters)
+	{
+        if(_fullVHostAccess)
+        {
+            //user has been granted full access to the vhost
+            return AuthzResult.ALLOWED;
+        }
+
+        Map createRights = (Map) _permissions.get(permission);
 
 		// If there are no create rights then deny request
 		if (createRights == null)
@@ -550,8 +594,15 @@ public class PrincipalPermissions
 		}
 	}
 
-	private AuthzResult authoriseBind(Object... parameters) {
-		Exchange exchange = (Exchange) parameters[1];
+	private AuthzResult authoriseBind(Object... parameters)
+	{
+        if(_fullVHostAccess)
+        {
+            //user has been granted full access to the vhost
+            return AuthzResult.ALLOWED;
+        }
+	    
+        Exchange exchange = (Exchange) parameters[1];
 
 		AMQQueue bind_queueName = (AMQQueue) parameters[2];
 		AMQShortString routingKey = (AMQShortString) parameters[3];
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/SimpleXML.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/SimpleXML.java
index b02e6b019c..532ff0b7d6 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/SimpleXML.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/SimpleXML.java
@@ -77,8 +77,29 @@ public class SimpleXML implements ACLPlugin
         processConsume(config);
 
         processCreate(config);
+        
+        processAccess(config);
     }
 
+    private void processAccess(Configuration config)
+    {
+        Configuration accessConfig = config.subset("access_control_list.access");
+        
+        if(accessConfig.isEmpty())
+        {
+            //there is no access configuration to process
+            return;
+        }
+        
+        // Process users that have full access permission
+        String[] users = accessConfig.getStringArray("users.user");
+
+        for (String user : users)
+        {
+            grant(Permission.ACCESS, user);
+        }
+    }
+    
     /**
      * Publish format takes Exchange + Routing Key Pairs
      *
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/security/access/PrincipalPermissionsTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/access/PrincipalPermissionsTest.java
index 28161a0229..e21edd1193 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/security/access/PrincipalPermissionsTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/access/PrincipalPermissionsTest.java
@@ -164,4 +164,37 @@ public class PrincipalPermissionsTest extends TestCase
         assertEquals(AuthzResult.ALLOWED, _perms.authorise(Permission.PUBLISH, authArgs));
     }
 
+    public void testVhostAccess()
+    {
+        //Tests that granting a user Virtualhost level access allows all authorisation requests
+        //where previously they would be denied 
+        
+        //QPID-2133 createExchange rights currently allow all exchange creation unless rights for creating some
+        //specific exchanges are granted. Grant a specific exchange creation to cause all others to be denied.
+        Object[] createArgsCreateExchange = new Object[]{new AMQShortString("madeup"), _exchangeType};
+        Object[] authArgsCreateExchange = new Object[]{_exchangeName,_exchangeType};
+        assertEquals("Exchange creation was not allowed", AuthzResult.ALLOWED, _perms.authorise(Permission.CREATEEXCHANGE, authArgsCreateExchange));
+        _perms.grant(Permission.CREATEEXCHANGE, createArgsCreateExchange);
+        
+        Object[] authArgsPublish = new Object[]{_exchange, _routingKey};        
+        Object[] authArgsConsume = new Object[]{_queue};
+        Object[] authArgsCreateQueue = new Object[]{_autoDelete, _queueName};
+        QueueBindBodyImpl bind = new QueueBindBodyImpl(_ticket, _queueName, _exchangeName, _routingKey, _nowait, _arguments);
+        Object[] authArgsBind = new Object[]{bind, _exchange, _queue, _routingKey};
+        
+        assertEquals("Exchange creation was not denied", AuthzResult.DENIED, _perms.authorise(Permission.CREATEEXCHANGE, authArgsCreateExchange));
+        assertEquals("Publish was not denied", AuthzResult.DENIED, _perms.authorise(Permission.PUBLISH, authArgsPublish));
+        assertEquals("Consume creation was not denied", AuthzResult.DENIED, _perms.authorise(Permission.CONSUME, authArgsConsume));
+        assertEquals("Queue creation was not denied", AuthzResult.DENIED, _perms.authorise(Permission.CREATEQUEUE, authArgsCreateQueue));
+        //BIND pre-grant authorise check disabled due to QPID-1597
+        //assertEquals("Binding creation was not denied", AuthzResult.DENIED, _perms.authorise(Permission.BIND, authArgsBind));
+        
+        _perms.grant(Permission.ACCESS);
+
+        assertEquals("Exchange creation was not allowed", AuthzResult.ALLOWED, _perms.authorise(Permission.CREATEEXCHANGE, authArgsCreateExchange));
+        assertEquals("Publish was not allowed", AuthzResult.ALLOWED, _perms.authorise(Permission.PUBLISH, authArgsPublish));
+        assertEquals("Consume creation was not allowed", AuthzResult.ALLOWED, _perms.authorise(Permission.CONSUME, authArgsConsume));
+        assertEquals("Queue creation was not allowed", AuthzResult.ALLOWED, _perms.authorise(Permission.CREATEQUEUE, authArgsCreateQueue));
+        assertEquals("Binding creation was not allowed", AuthzResult.ALLOWED, _perms.authorise(Permission.BIND, authArgsBind));
+    }
 }
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java b/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java
index 79bc29a4e3..6b88b6a22d 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java
@@ -23,6 +23,7 @@ package org.apache.qpid.client.failover;
 import org.apache.qpid.AMQDisconnectedException;
 import org.apache.qpid.client.protocol.AMQProtocolHandler;
 import org.apache.qpid.client.state.AMQStateManager;
+import org.apache.qpid.client.state.AMQState;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -193,6 +194,22 @@ public class FailoverHandler implements Runnable
                 // Set the new Protocol Session in the StateManager.               
                 existingStateManager.setProtocolSession(_amqProtocolHandler.getProtocolSession());
 
+                // Now that the ProtocolHandler has been reconnected clean up
+                // the state of the old state manager. As if we simply reinstate
+                // it any old exception that had occured prior to failover may
+                // prohibit reconnection.
+                // e.g. During testing when the broker is shutdown gracefully.
+                // The broker 
+                // Clear any exceptions we gathered
+                if (existingStateManager.getCurrentState() != AMQState.CONNECTION_OPEN)
+                {
+                    // Clear the state of the previous state manager as it may
+                    // have received an exception
+                    existingStateManager.clearLastException();
+                    existingStateManager.changeState(AMQState.CONNECTION_OPEN);
+                }
+
+
                 //Restore Existing State Manager
                 _amqProtocolHandler.setStateManager(existingStateManager);
                 try
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java b/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java
index 70d4697f2c..9c7d62670c 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java
@@ -213,4 +213,9 @@ public class AMQStateManager implements AMQMethodListener
     {
         return _lastException;
     }
+
+    public void clearLastException()
+    {
+        _lastException = null;
+    }
 }
diff --git a/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverRoundRobinServers.java b/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverRoundRobinServers.java
index 869944dd6e..41ba4974ec 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverRoundRobinServers.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/jms/failover/FailoverRoundRobinServers.java
@@ -95,11 +95,10 @@ public class FailoverRoundRobinServers implements FailoverMethod
 
     public boolean failoverAllowed()
     {
-        System.out.println("====================================");
-        System.out.println(toString());
-        System.out.println("====================================");
+        _logger.info("==== Checking failoverAllowed() ====");
+        _logger.info(toString());
+        _logger.info("====================================");
         return ((_currentCycleRetries < _cycleRetries) || (_currentServerRetry < _serverRetries));
-                //|| (_currentBrokerIndex <= (_connectionDetails.getBrokerCount() - 1)));
     }
 
     public void attainedConnection()
@@ -175,7 +174,11 @@ public class FailoverRoundRobinServers implements FailoverMethod
         }
         else
         {
-            _logger.info("No delay between connect retries, use tcp://host:port?connectdelay='value' to enable.");
+            // Only display if option not set. Not if deDelay==false.
+            if (delayStr == null)
+            {
+                _logger.info("No delay between connect retries, use tcp://host:port?connectdelay='value' to enable.");
+            }
         }
 
         return broker;
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/SimpleACLTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/SimpleACLTest.java
index 6e1d68efa1..04d1ec1e72 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/SimpleACLTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/security/acl/SimpleACLTest.java
@@ -56,6 +56,20 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener, E
     	//Each test now calls the private setUpACLTest to allow them to make 
     	//individual customisations to the base ACL settings
     }
+
+
+    public void tearDown() throws Exception
+    {
+        try
+        {
+            super.tearDown();
+        }
+        catch (JMSException e)
+        {
+            //we're throwing this away as it can happen in this test as the state manager remembers exceptions
+            //that we provoked with authentication failures, where the test passes - we can ignore on con close
+        }
+    }
     
     private void setUpACLTest() throws Exception
     {
@@ -100,7 +114,51 @@ public class SimpleACLTest extends QpidTestCase implements ConnectionListener, E
             fail("Connection was not created due to:" + e);
         }
     }
+    
+    public void testAccessVhostAuthorisedGuest() throws IOException, Exception
+    {
+        //The 'guest' user normally has no access, as tested below in testAccessNoRights(), and so is unable to perform
+        //actions such as connecting (and by extension, creating a queue, and consuming from a queue etc). In order to test
+        //the vhost-wide 'access' right, we will now give the guest user 'access' ACL rights and perform various such actions.
+        setConfigurationProperty("virtualhosts.virtualhost.test.security.access_control_list.access.users.user", "guest");
+
+        setUpACLTest();
+        
+        try
+        {
+            //get a connection
+            Connection conn = getConnection("guest", "guest");
+            ((AMQConnection) conn).setConnectionListener(this);
+
+            Session sesh = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+            conn.start();
+
+            //create Queues and consumers for each
+            Queue namedQueue = sesh.createQueue("vhostAccessCreatedQueue" + getTestQueueName());
+            Queue tempQueue = sesh.createTemporaryQueue();
+            MessageConsumer consumer = sesh.createConsumer(namedQueue);
+            MessageConsumer tempConsumer = sesh.createConsumer(tempQueue);
+
+            //send a message to each queue (also causing an exchange declare)
+            MessageProducer sender = ((AMQSession)sesh).createProducer(null);
+            ((org.apache.qpid.jms.MessageProducer) sender).send(namedQueue, sesh.createTextMessage("test"),
+                                                                DeliveryMode.NON_PERSISTENT, 0, 0L, false, false, true);
+            ((org.apache.qpid.jms.MessageProducer) sender).send(tempQueue, sesh.createTextMessage("test"),
+                                                                DeliveryMode.NON_PERSISTENT, 0, 0L, false, false, true);
 
+            //consume the messages from the queues
+            consumer.receive(2000);
+            tempConsumer.receive(2000);
+
+            conn.close();
+        }
+        catch (Exception e)
+        {
+            fail("Test failed due to:" + e.getMessage());
+        }
+    }
+    
     public void testAccessNoRights() throws Exception
     {
     	setUpACLTest();