QPID-3310 - Principal/Subject refactoring.

Refactoring to the connection/session objects to pass the Subject from Authentication tier to Access tier, rather than just
the Principal.  Change the access-control to be able to make access decisions based on Groups from the Authentication tier
whilst retaining support for groups declared within the ACL file itself.  Improve unit tests.

Applied patch by Keith Wall <keith.wall@gmail.com>

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

37 files changed, 1198 insertions, 548 deletions

diff --git a/java/broker-plugins/access-control/MANIFEST.MF b/java/broker-plugins/access-control/MANIFEST.MF
index 1cd285ba20..78072850e4 100644
--- a/java/broker-plugins/access-control/MANIFEST.MF
+++ b/java/broker-plugins/access-control/MANIFEST.MF
@@ -35,6 +35,7 @@ Import-Package: org.apache.qpid,
  org.apache.log4j;version=1.0.0,
  javax.management;version=1.0.0,
  javax.management.openmbean;version=1.0.0,
+ javax.security.auth;version=1.0.0,
  org.osgi.util.tracker;version=1.0.0,
  org.osgi.framework;version=1.3
 Private-Package: org.apache.qpid.server.security.access.config,
diff --git a/java/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/RuleSet.java b/java/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/RuleSet.java
index ebc73440ed..78355a7501 100644
--- a/java/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/RuleSet.java
+++ b/java/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/config/RuleSet.java
@@ -18,20 +18,24 @@
  */
 package org.apache.qpid.server.security.access.config;
 
+import java.security.Principal;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.EnumMap;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.SortedMap;
 import java.util.TreeMap;
 import java.util.WeakHashMap;
 
+import javax.security.auth.Subject;
+
 import org.apache.commons.lang.BooleanUtils;
 import org.apache.commons.lang.StringUtils;
-import org.apache.log4j.Logger;
 import org.apache.qpid.exchange.ExchangeDefaults;
 import org.apache.qpid.server.logging.actors.CurrentActor;
 import org.apache.qpid.server.security.Result;
@@ -45,147 +49,132 @@ import org.apache.qpid.server.security.access.logging.AccessControlMessages;
  * Models the rule configuration for the access control plugin.
  *
  * The access control rule definitions are loaded from an external configuration file, passed in as the
- * target to the {@link load(ConfigurationFile)} method. The file specified 
+ * target to the {@link load(ConfigurationFile)} method. The file specified
  */
 public class RuleSet
 {
-    private static final Logger _logger = Logger.getLogger(RuleSet.class);
-    
     private static final String AT = "@";
-	private static final String SLASH = "/";
+    private static final String SLASH = "/";
 
-	public static final String DEFAULT_ALLOW = "defaultallow";
-	public static final String DEFAULT_DENY = "defaultdeny";
-	public static final String TRANSITIVE = "transitive";
-	public static final String EXPAND = "expand";
+    public static final String DEFAULT_ALLOW = "defaultallow";
+    public static final String DEFAULT_DENY = "defaultdeny";
+    public static final String TRANSITIVE = "transitive";
+    public static final String EXPAND = "expand";
     public static final String AUTONUMBER = "autonumber";
     public static final String CONTROLLED = "controlled";
     public static final String VALIDATE = "validate";
-    
+
     public static final List<String> CONFIG_PROPERTIES = Arrays.asList(
             DEFAULT_ALLOW, DEFAULT_DENY, TRANSITIVE, EXPAND, AUTONUMBER, CONTROLLED
         );
-    
+
     private static final Integer _increment = 10;
-	
-    private final Map<String, List<String>> _groups = new HashMap<String, List<String>>();
+
+    private final Map<String, List<String>> _aclGroups = new HashMap<String, List<String>>();
     private final SortedMap<Integer, Rule> _rules = new TreeMap<Integer, Rule>();
-    private final Map<String, Map<Operation, Map<ObjectType, List<Rule>>>> _cache =
-                        new WeakHashMap<String, Map<Operation, Map<ObjectType, List<Rule>>>>();
+    private final Map<Subject, Map<Operation, Map<ObjectType, List<Rule>>>> _cache =
+                        new WeakHashMap<Subject, Map<Operation, Map<ObjectType, List<Rule>>>>();
     private final Map<String, Boolean> _config = new HashMap<String, Boolean>();
-    
+
     public RuleSet()
     {
         // set some default configuration properties
         configure(DEFAULT_DENY, Boolean.TRUE);
         configure(TRANSITIVE, Boolean.TRUE);
     }
-    
+
     /**
-     * Clear the contents, invluding groups, rules and configuration.
+     * Clear the contents, including acl groups, rules and configuration.
      */
     public void clear()
     {
         _rules.clear();
         _cache.clear();
         _config.clear();
-        _groups.clear();
+        _aclGroups.clear();
     }
-    
+
     public int getRuleCount()
     {
         return _rules.size();
     }
-	
-	/**
-	 * Filtered rules list based on an identity and operation.
-	 * 
-	 * Allows only enabled rules with identity equal to all, the same, or a group with identity as a member,
-	 * and operation is either all or the same operation.
-	 */		
-	public List<Rule> getRules(String identity, Operation operation, ObjectType objectType)
-	{
-        // Lookup identity in cache and create empty operation map if required
-		Map<Operation, Map<ObjectType, List<Rule>>> operations = _cache.get(identity);		
-		if (operations == null)
-		{	
-			operations = new EnumMap<Operation, Map<ObjectType, List<Rule>>>(Operation.class);
-			_cache.put(identity, operations);
-		}
-		
-        // Lookup operation and create empty object type map if required        
-        Map<ObjectType, List<Rule>> objects = operations.get(operation);
-		if (objects == null)
-		{
-            objects = new EnumMap<ObjectType, List<Rule>>(ObjectType.class);
-            operations.put(operation, objects);
-        }
+
+    /**
+     * Filtered rules list based on a subject and operation.
+     *
+     * Allows only enabled rules with identity equal to all, the same, or a group with identity as a member,
+     * and operation is either all or the same operation.
+     */
+    public List<Rule> getRules(final Subject subject, final Operation operation, final ObjectType objectType)
+    {
+        final Map<ObjectType, List<Rule>> objects = getObjectToRuleCache(subject, operation);
 
         // Lookup object type rules for the operation
         if (!objects.containsKey(objectType))
         {
+            final Set<Principal> principals = subject.getPrincipals();
             boolean controlled = false;
             List<Rule> filtered = new LinkedList<Rule>();
             for (Rule rule : _rules.values())
             {
+                final Action ruleAction = rule.getAction();
                 if (rule.isEnabled()
-                    && (rule.getAction().getOperation() == Operation.ALL || rule.getAction().getOperation() == operation)
-                    && (rule.getAction().getObjectType() == ObjectType.ALL || rule.getAction().getObjectType() == objectType))
+                    && (ruleAction.getOperation() == Operation.ALL || ruleAction.getOperation() == operation)
+                    && (ruleAction.getObjectType() == ObjectType.ALL || ruleAction.getObjectType() == objectType))
                 {
                     controlled = true;
 
-                    if (rule.getIdentity().equalsIgnoreCase(Rule.ALL)
-                        || rule.getIdentity().equalsIgnoreCase(identity)
-                        || (_groups.containsKey(rule.getIdentity()) && _groups.get(rule.getIdentity()).contains(identity)))
+                    if (isRelevant(principals,rule))
                     {
                         filtered.add(rule);
                     }
                 }
             }
-            
+
             // Return null if there are no rules at all for this operation and object type
             if (filtered.isEmpty() && controlled == false)
             {
                 filtered = null;
             }
-            
+
             // Save the rules we selected
             objects.put(objectType, filtered);
         }
-		
+
         // Return the cached rules
-		return objects.get(objectType);
-	}
-    
+        return objects.get(objectType);
+    }
+
+
     public boolean isValidNumber(Integer number)
     {
         return !_rules.containsKey(number);
     }
-	
+
     public void grant(Integer number, String identity, Permission permission, Operation operation)
     {
         Action action = new Action(operation);
         addRule(number, identity, permission, action);
     }
-    
+
     public void grant(Integer number, String identity, Permission permission, Operation operation, ObjectType object, ObjectProperties properties)
     {
         Action action = new Action(operation, object, properties);
         addRule(number, identity, permission, action);
     }
-    
+
     public boolean ruleExists(String identity, Action action)
     {
-		for (Rule rule : _rules.values())
-		{
-		    if (rule.getIdentity().equals(identity) && rule.getAction().equals(action))
-		    {
-		        return true;
-		    }
-		}
-		return false;
+        for (Rule rule : _rules.values())
+        {
+            if (rule.getIdentity().equals(identity) && rule.getAction().equals(action))
+            {
+                return true;
+            }
+        }
+        return false;
     }
-    
+
     private Permission noLog(Permission permission)
     {
         switch (permission)
@@ -203,15 +192,17 @@ public class RuleSet
     // TODO make this work when group membership is not known at file parse time
     public void addRule(Integer number, String identity, Permission permission, Action action)
     {
-		if (!action.isAllowed())
-		{
-			throw new IllegalArgumentException("Action is not allowd: " + action);
-		}
+        _cache.clear();
+
+        if (!action.isAllowed())
+        {
+            throw new IllegalArgumentException("Action is not allowd: " + action);
+        }
         if (ruleExists(identity, action))
         {
             return;
         }
-        
+
         // expand actions - possibly multiply number by
         if (isSet(EXPAND))
         {
@@ -234,8 +225,8 @@ public class RuleSet
                 return;
             }
         }
-        
-		// transitive action dependencies
+
+        // transitive action dependencies
         if (isSet(TRANSITIVE))
         {
             if (action.getOperation() == Operation.CREATE && action.getObjectType() == ObjectType.QUEUE)
@@ -244,10 +235,10 @@ public class RuleSet
                 exchProperties.setName(ExchangeDefaults.DEFAULT_EXCHANGE_NAME);
                 exchProperties.put(ObjectProperties.Property.ROUTING_KEY, action.getProperties().get(ObjectProperties.Property.NAME));
                 addRule(null, identity, noLog(permission), new Action(Operation.BIND, ObjectType.EXCHANGE, exchProperties));
-				if (action.getProperties().isSet(ObjectProperties.Property.AUTO_DELETE))
-				{
-					addRule(null, identity, noLog(permission), new Action(Operation.DELETE, ObjectType.QUEUE, action.getProperties()));
-				}
+                if (action.getProperties().isSet(ObjectProperties.Property.AUTO_DELETE))
+                {
+                    addRule(null, identity, noLog(permission), new Action(Operation.DELETE, ObjectType.QUEUE, action.getProperties()));
+                }
             }
             else if (action.getOperation() == Operation.DELETE && action.getObjectType() == ObjectType.QUEUE)
             {
@@ -261,9 +252,9 @@ public class RuleSet
                 addRule(null, identity, noLog(permission), new Action(Operation.ACCESS, ObjectType.VIRTUALHOST));
             }
         }
-        
+
         // set rule number if needed
-        Rule rule = new Rule(number, identity, action, permission);        
+        Rule rule = new Rule(number, identity, action, permission);
         if (rule.getNumber() == null)
         {
             if (_rules.isEmpty())
@@ -275,34 +266,36 @@ public class RuleSet
                 rule.setNumber(_rules.lastKey() + _increment);
             }
         }
-        
+
         // save rule
         _cache.remove(identity);
         _rules.put(rule.getNumber(), rule);
-	}
-    
+    }
+
     public void enableRule(int ruleNumber)
     {
         _rules.get(Integer.valueOf(ruleNumber)).enable();
     }
-    
+
     public void disableRule(int ruleNumber)
     {
         _rules.get(Integer.valueOf(ruleNumber)).disable();
     }
-    
+
     public boolean addGroup(String group, List<String> constituents)
     {
-        if (_groups.containsKey(group))
+        _cache.clear();
+
+        if (_aclGroups.containsKey(group))
         {
             // cannot redefine
             return false;
         }
         else
         {
-            _groups.put(group, new ArrayList<String>());
+            _aclGroups.put(group, new ArrayList<String>());
         }
-        
+
         for (String name : constituents)
         {
             if (name.equalsIgnoreCase(group))
@@ -310,17 +303,17 @@ public class RuleSet
                 // recursive definition
                 return false;
             }
-            
+
             if (!checkName(name))
             {
                 // invalid name
                 return false;
             }
-            
-            if (_groups.containsKey(name))
+
+            if (_aclGroups.containsKey(name))
             {
                 // is a group
-                _groups.get(group).addAll(_groups.get(name));
+                _aclGroups.get(group).addAll(_aclGroups.get(name));
             }
             else
             {
@@ -330,12 +323,12 @@ public class RuleSet
                     // invalid username
                     return false;
                 }
-                _groups.get(group).add(name);
+                _aclGroups.get(group).add(name);
             }
         }
         return true;
     }
-    
+
     /** Return true if the name is well-formed (contains legal characters). */
     protected boolean checkName(String name)
     {
@@ -349,79 +342,79 @@ public class RuleSet
         }
         return true;
     }
-    
+
     /** Returns true if a username has the name[@domain][/realm] format  */
     protected boolean isvalidUserName(String name)
-    {	
-		// check for '@' and '/' in namne
-		int atPos = name.indexOf(AT);
-		int slashPos = name.indexOf(SLASH);
-		boolean atFound = atPos != StringUtils.INDEX_NOT_FOUND && atPos == name.lastIndexOf(AT);
-		boolean slashFound = slashPos != StringUtils.INDEX_NOT_FOUND && slashPos == name.lastIndexOf(SLASH);
-				
-		// must be at least one character after '@' or '/'
-		if (atFound && atPos > name.length() - 2)
-		{
-			return false;
-		}
-		if (slashFound && slashPos > name.length() - 2)
-		{
-			return false;
-		}
-		
-		// must be at least one character between '@' and '/'
-		if (atFound && slashFound)
-		{
-			return (atPos < (slashPos - 1));
-		}
-		
-		// otherwise all good
-		return true; 
+    {
+        // check for '@' and '/' in namne
+        int atPos = name.indexOf(AT);
+        int slashPos = name.indexOf(SLASH);
+        boolean atFound = atPos != StringUtils.INDEX_NOT_FOUND && atPos == name.lastIndexOf(AT);
+        boolean slashFound = slashPos != StringUtils.INDEX_NOT_FOUND && slashPos == name.lastIndexOf(SLASH);
+
+        // must be at least one character after '@' or '/'
+        if (atFound && atPos > name.length() - 2)
+        {
+            return false;
+        }
+        if (slashFound && slashPos > name.length() - 2)
+        {
+            return false;
+        }
+
+        // must be at least one character between '@' and '/'
+        if (atFound && slashFound)
+        {
+            return (atPos < (slashPos - 1));
+        }
+
+        // otherwise all good
+        return true;
     }
 
-	// C++ broker authorise function prototype
+    // C++ broker authorise function prototype
     // virtual bool authorise(const std::string& id, const Action& action, const ObjectType& objType,
-	//		const std::string& name, std::map<Property, std::string>* params=0);
-	
-	// Possibly add a String name paramater?
+    //        const std::string& name, std::map<Property, std::string>* params=0);
+
+    // Possibly add a String name paramater?
 
     /**
      * Check the authorisation granted to a particular identity for an operation on an object type with
      * specific properties.
      *
-     * Looks up the entire ruleset, whcih may be cached, for the user and operation and goes through the rules
+     * Looks up the entire ruleset, which may be cached, for the user and operation and goes through the rules
      * in order to find the first one that matches. Either defers if there are no rules, returns the result of
      * the first match found, or denies access if there are no matching rules. Normally, it would be expected
      * to have a default deny or allow rule at the end of an access configuration however.
      */
-    public Result check(String identity, Operation operation, ObjectType objectType, ObjectProperties properties)
+    public Result check(Subject subject, Operation operation, ObjectType objectType, ObjectProperties properties)
     {
         // Create the action to check
         Action action = new Action(operation, objectType, properties);
 
-		// get the list of rules relevant for this request
-		List<Rule> rules = getRules(identity, operation, objectType);
-		if (rules == null)
-		{
-		    if (isSet(CONTROLLED))
-		    {
-    		    // Abstain if there are no rules for this operation
+        // get the list of rules relevant for this request
+        List<Rule> rules = getRules(subject, operation, objectType);
+        if (rules == null)
+        {
+            if (isSet(CONTROLLED))
+            {
+                // Abstain if there are no rules for this operation
                 return Result.ABSTAIN;
-		    }
-		    else
-		    {
-		        return getDefault();
-		    }
-		}
-		
-		// Iterate through a filtered set of rules dealing with this identity and operation
+            }
+            else
+            {
+                return getDefault();
+            }
+        }
+
+        // Iterate through a filtered set of rules dealing with this identity and operation
         for (Rule current : rules)
-		{
-			// Check if action matches
+        {
+            // Check if action matches
             if (action.matches(current.getAction()))
             {
                 Permission permission = current.getPermission();
-                
+
                 switch (permission)
                 {
                     case ALLOW_LOG:
@@ -439,15 +432,15 @@ public class RuleSet
                 return Result.DENIED;
             }
         }
-        
+
         // Defer to the next plugin of this type, if it exists
-		return Result.DEFER;
+        return Result.DEFER;
     }
-	
-	/** Default deny. */
-	public Result getDefault()
-	{
-	    if (isSet(DEFAULT_ALLOW))
+
+    /** Default deny. */
+    public Result getDefault()
+    {
+        if (isSet(DEFAULT_ALLOW))
         {
             return Result.ALLOWED;
         }
@@ -456,19 +449,19 @@ public class RuleSet
             return Result.DENIED;
         }
         return Result.ABSTAIN;
-	}
-	
-	/**
-	 * Check if a configuration property is set.
-	 */
-	protected boolean isSet(String key)
-	{
-	    return BooleanUtils.isTrue(_config.get(key));
-	}
+    }
+
+    /**
+     * Check if a configuration property is set.
+     */
+    protected boolean isSet(String key)
+    {
+        return BooleanUtils.isTrue(_config.get(key));
+    }
 
     /**
      * Configure properties for the plugin instance.
-     * 
+     *
      * @param properties
      */
     public void configure(Map<String, Boolean> properties)
@@ -478,7 +471,7 @@ public class RuleSet
 
     /**
      * Configure a single property for the plugin instance.
-     * 
+     *
      * @param key
      * @param value
      */
@@ -486,4 +479,48 @@ public class RuleSet
     {
         _config.put(key, value);
     }
+
+    private boolean isRelevant(final Set<Principal> principals, final Rule rule)
+    {
+        if (rule.getIdentity().equalsIgnoreCase(Rule.ALL))
+        {
+            return true;
+        }
+        else
+        {
+            for (Iterator<Principal> iterator = principals.iterator(); iterator.hasNext();)
+            {
+                final Principal principal = iterator.next();
+
+                if (rule.getIdentity().equalsIgnoreCase(principal.getName())
+                    || (_aclGroups.containsKey(rule.getIdentity()) && _aclGroups.get(rule.getIdentity()).contains(principal.getName())))
+                {
+                    return true;
+                }
+            }
+        }
+
+        return false;
+    }
+
+    private Map<ObjectType, List<Rule>> getObjectToRuleCache(final Subject subject, final Operation operation)
+    {
+        // Lookup identity in cache and create empty operation map if required
+        Map<Operation, Map<ObjectType, List<Rule>>> operations = _cache.get(subject);
+        if (operations == null)
+        {
+            operations = new EnumMap<Operation, Map<ObjectType, List<Rule>>>(Operation.class);
+            _cache.put(subject, operations);
+        }
+
+        // Lookup operation and create empty object type map if required
+        Map<ObjectType, List<Rule>> objects = operations.get(operation);
+        if (objects == null)
+        {
+            objects = new EnumMap<ObjectType, List<Rule>>(ObjectType.class);
+            operations.put(operation, objects);
+        }
+        return objects;
+    }
+
 }
diff --git a/java/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/plugins/AccessControl.java b/java/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/plugins/AccessControl.java
index 69cfa173bd..a7b3059262 100644
--- a/java/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/plugins/AccessControl.java
+++ b/java/broker-plugins/access-control/src/main/java/org/apache/qpid/server/security/access/plugins/AccessControl.java
@@ -20,7 +20,7 @@
  */
 package org.apache.qpid.server.security.access.plugins;
 
-import java.security.Principal;
+import javax.security.auth.Subject;
 
 import org.apache.commons.configuration.ConfigurationException;
 import org.apache.log4j.Logger;
@@ -89,20 +89,19 @@ public class AccessControl extends AbstractPlugin
     
     /**
      * Check if an operation is authorised by asking the  configuration object about the access
-     * control rules granted to the current thread's {@link Principal}. If there is no current
+     * control rules granted to the current thread's {@link Subject}. If there is no current
      * user the plugin will abstain.
      */
     public Result authorise(Operation operation, ObjectType objectType, ObjectProperties properties)
     {
-        Principal principal = SecurityManager.getThreadPrincipal();
-        
-        // Abstain if there is no user associated with this thread
-        if (principal == null)
+        final Subject subject = SecurityManager.getThreadSubject();
+        // Abstain if there is no subject/principal associated with this thread
+        if (subject == null  || subject.getPrincipals().size() == 0)
         {
             return Result.ABSTAIN;
         }
-        
-        return _ruleSet.check(principal.getName(), operation, objectType, properties);
+ 
+        return  _ruleSet.check(subject, operation, objectType, properties);
     }
 
     public void configure(ConfigurationPlugin config)
diff --git a/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/AccessControlTest.java b/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/AccessControlTest.java
index 309a3aeb2c..09d26e5451 100644
--- a/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/AccessControlTest.java
+++ b/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/AccessControlTest.java
@@ -1,195 +1,172 @@
 /*
- *  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
+ * 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.
  *
- *  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.security.access.plugins;
 
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.FileWriter;
-import java.io.PrintWriter;
+import java.util.Arrays;
 
 import junit.framework.TestCase;
 
-import org.apache.commons.configuration.ConfigurationException;
-import org.apache.qpid.server.security.access.config.ConfigurationFile;
-import org.apache.qpid.server.security.access.config.PlainConfiguration;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
+import org.apache.qpid.server.logging.UnitTestMessageLogger;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.TestLogActor;
+import org.apache.qpid.server.security.Result;
+import org.apache.qpid.server.security.SecurityManager;
+import org.apache.qpid.server.security.access.ObjectProperties;
+import org.apache.qpid.server.security.access.ObjectType;
+import org.apache.qpid.server.security.access.Operation;
+import org.apache.qpid.server.security.access.Permission;
+import org.apache.qpid.server.security.access.config.Rule;
 import org.apache.qpid.server.security.access.config.RuleSet;
+import org.apache.qpid.server.security.auth.sasl.TestPrincipalUtils;
 
 /**
- * These tests check that the ACL file parsing works correctly.
+ * Unit test for ACL V2 plugin.  
+ * 
+ * This unit test tests the AccessControl class and it collaboration with {@link RuleSet},
+ * {@link SecurityManager} and {@link CurrentActor}.   The ruleset is configured programmatically,
+ * rather than from an external file.
  * 
- * For each message that can be returned in a {@link ConfigurationException}, an ACL file is created that should trigger this
- * particular message.
+ * @see RuleSetTest
  */
 public class AccessControlTest extends TestCase
 {
-    public void writeACLConfig(String...aclData) throws Exception
+    private AccessControl _plugin = null;  // Class under test
+    private final UnitTestMessageLogger messageLogger = new UnitTestMessageLogger();
+
+    protected void setUp() throws Exception
     {
-        File acl = File.createTempFile(getClass().getName() + getName(), "acl");
-        acl.deleteOnExit();
-        
-        // Write ACL file
-        PrintWriter aclWriter = new PrintWriter(new FileWriter(acl));
-        for (String line : aclData)
-        {
-            aclWriter.println(line);
-        }
-        aclWriter.close();
+        super.setUp();
 
-        // Load ruleset
-        ConfigurationFile configFile = new PlainConfiguration(acl);
-        RuleSet ruleSet = configFile.load();
-    }
+        final RuleSet rs = new RuleSet();
+        rs.addGroup("aclGroup1", Arrays.asList(new String[] {"member1", "member2"}));
 
-    public void testMissingACLConfig() throws Exception
-    {
-        try
-        {
-            // Load ruleset
-	        ConfigurationFile configFile = new PlainConfiguration(new File("doesnotexist"));
-	        RuleSet ruleSet = configFile.load();
-            
-            fail("fail");
-        }
-        catch (ConfigurationException ce)
-        {
-            assertEquals(String.format(PlainConfiguration.CONFIG_NOT_FOUND_MSG, "doesnotexist"), ce.getMessage());
-            assertTrue(ce.getCause() instanceof FileNotFoundException);
-            assertEquals("doesnotexist (No such file or directory)", ce.getCause().getMessage());
-        }
-    }
+        // Rule expressed with username
+        rs.grant(0, "user1", Permission.ALLOW, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        // Rule expressed with a acl group
+        rs.grant(1, "aclGroup1", Permission.ALLOW, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        // Rule expressed with an external group
+        rs.grant(2, "extGroup1", Permission.DENY, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        // Catch all rule
+        rs.grant(3, Rule.ALL, Permission.DENY_LOG, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
 
-    public void testACLFileSyntaxContinuation() throws Exception
-    {
-        try
-        {
-            writeACLConfig("ACL ALLOW ALL \\ ALL");
-            fail("fail");
-        }
-        catch (ConfigurationException ce)
-        {
-            assertEquals(String.format(PlainConfiguration.PREMATURE_CONTINUATION_MSG, 1), ce.getMessage());
-        }
-    }
+        _plugin = (AccessControl) AccessControl.FACTORY.newInstance(createConfiguration(rs));
 
-    public void testACLFileSyntaxTokens() throws Exception
-    {
-        try
-        {
-            writeACLConfig("ACL unparsed ALL ALL");
-            fail("fail");
-        }
-        catch (ConfigurationException ce)
-        {
-            assertEquals(String.format(PlainConfiguration.PARSE_TOKEN_FAILED_MSG, 1), ce.getMessage());
-            assertTrue(ce.getCause() instanceof IllegalArgumentException);
-            assertEquals("Not a valid permission: unparsed", ce.getCause().getMessage());
-        }
+        SecurityManager.setThreadSubject(null);
+        
+        CurrentActor.set(new TestLogActor(messageLogger));
     }
 
-    public void testACLFileSyntaxNotEnoughGroup() throws Exception
+    protected void tearDown() throws Exception
     {
-        try
-        {
-            writeACLConfig("GROUP blah");
-            fail("fail");
-        }
-        catch (ConfigurationException ce)
-        {
-            assertEquals(String.format(PlainConfiguration.NOT_ENOUGH_GROUP_MSG, 1), ce.getMessage());
-        }
+        super.tearDown();
+        SecurityManager.setThreadSubject(null);
     }
 
-    public void testACLFileSyntaxNotEnoughACL() throws Exception
+    /** 
+     * ACL plugin must always abstain if there is no  subject attached to the thread.
+     */
+    public void testNoSubjectAlwaysAbstains()
     {
-        try
-        {
-            writeACLConfig("ACL ALLOW");
-            fail("fail");
-        }
-        catch (ConfigurationException ce)
-        {
-            assertEquals(String.format(PlainConfiguration.NOT_ENOUGH_ACL_MSG, 1), ce.getMessage());
-        }
+        SecurityManager.setThreadSubject(null);
+
+        final Result result = _plugin.authorise(Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        assertEquals(Result.ABSTAIN, result);
     }
 
-    public void testACLFileSyntaxNotEnoughConfig() throws Exception
+    /** 
+     * Tests that an allow rule expressed with a username allows an operation performed by a thread running
+     * with the same username.
+     */
+    public void testUsernameAllowsOperation()
     {
-        try
-        {
-            writeACLConfig("CONFIG");
-            fail("fail");
-        }
-        catch (ConfigurationException ce)
-        {
-            assertEquals(String.format(PlainConfiguration.NOT_ENOUGH_TOKENS_MSG, 1), ce.getMessage());
-        }
+        SecurityManager.setThreadSubject(TestPrincipalUtils.createTestSubject("user1"));
+
+        final Result result = _plugin.authorise(Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        assertEquals(Result.ALLOWED, result);
     }
 
-    public void testACLFileSyntaxNotEnough() throws Exception
+    /** 
+     * Tests that an allow rule expressed with an <b>ACL groupname</b> allows an operation performed by a thread running
+     * by a user who belongs to the same group..
+     */
+    public void testAclGroupMembershipAllowsOperation()
     {
-        try
-        {
-            writeACLConfig("INVALID");
-            fail("fail");
-        }
-        catch (ConfigurationException ce)
-        {
-            assertEquals(String.format(PlainConfiguration.NOT_ENOUGH_TOKENS_MSG, 1), ce.getMessage());
-        }
+        SecurityManager.setThreadSubject(TestPrincipalUtils.createTestSubject("member1"));
+
+        final Result result = _plugin.authorise(Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        assertEquals(Result.ALLOWED, result);
     }
 
-    public void testACLFileSyntaxPropertyKeyOnly() throws Exception
+    /** 
+     * Tests that a deny rule expressed with an <b>External groupname</b> denies an operation performed by a thread running
+     * by a user who belongs to the same group.
+     */
+    public void testExternalGroupMembershipDeniesOperation()
     {
-        try
-        {
-            writeACLConfig("ACL ALLOW adk CREATE QUEUE name");
-            fail("fail");
-        }
-        catch (ConfigurationException ce)
-        {
-            assertEquals(String.format(PlainConfiguration.PROPERTY_KEY_ONLY_MSG, 1), ce.getMessage());
-        }
+        SecurityManager.setThreadSubject(TestPrincipalUtils.createTestSubject("user3", "extGroup1"));
+        
+        final Result result = _plugin.authorise(Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        assertEquals(Result.DENIED, result);
     }
 
-    public void testACLFileSyntaxPropertyNoEquals() throws Exception
+    /** 
+     * Tests that the catch all deny denies the operation and logs with the logging actor.
+     */
+    public void testCatchAllRuleDeniesUnrecognisedUsername()
     {
-        try
-        {
-            writeACLConfig("ACL ALLOW adk CREATE QUEUE name test");
-            fail("fail");
-        }
-        catch (ConfigurationException ce)
-        {
-            assertEquals(String.format(PlainConfiguration.PROPERTY_NO_EQUALS_MSG, 1), ce.getMessage());
-        }
+        SecurityManager.setThreadSubject(TestPrincipalUtils.createTestSubject("unknown", "unkgroup1", "unkgroup2"));
+        
+        assertEquals("Expecting zero messages before test", 0, messageLogger.getLogMessages().size());
+        final Result result = _plugin.authorise(Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        assertEquals(Result.DENIED, result);
+        
+        assertEquals("Expecting one message before test", 1, messageLogger.getLogMessages().size());
+        assertTrue("Logged message does not contain expected string", messageLogger.messageContains(0, "ACL-1002"));
     }
-
-    public void testACLFileSyntaxPropertyNoValue() throws Exception
+    
+    /**
+     * Creates a configuration plugin for the {@link AccessControl} plugin.
+     */
+    private ConfigurationPlugin createConfiguration(final RuleSet rs)
     {
-        try
-        {
-            writeACLConfig("ACL ALLOW adk CREATE QUEUE name =");
-            fail("fail");
-        }
-        catch (ConfigurationException ce)
-        {
-            assertEquals(String.format(PlainConfiguration.PROPERTY_NO_VALUE_MSG, 1), ce.getMessage());
-        }
+        final ConfigurationPlugin cp = new ConfigurationPlugin()
+        {
+            public AccessControlConfiguration  getConfiguration(final String plugin)
+            {
+                return new AccessControlConfiguration()
+                {
+                    public RuleSet getRuleSet()
+                    {
+                        return rs;
+                    }
+                };
+            }
+
+            public String[] getElementsProcessed()
+            {
+                throw new UnsupportedOperationException();
+            }
+        };
+
+        return cp;
     }
 }
diff --git a/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/PlainConfigurationTest.java b/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/PlainConfigurationTest.java
new file mode 100644
index 0000000000..e16f9943ba
--- /dev/null
+++ b/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/PlainConfigurationTest.java
@@ -0,0 +1,194 @@
+/*
+ *  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.security.access.plugins;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileWriter;
+import java.io.PrintWriter;
+
+import junit.framework.TestCase;
+
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.qpid.server.security.access.config.ConfigurationFile;
+import org.apache.qpid.server.security.access.config.PlainConfiguration;
+
+/**
+ * These tests check that the ACL file parsing works correctly.
+ * 
+ * For each message that can be returned in a {@link ConfigurationException}, an ACL file is created that should trigger this
+ * particular message.
+ */
+public class PlainConfigurationTest extends TestCase
+{
+    public void writeACLConfig(String...aclData) throws Exception
+    {
+        File acl = File.createTempFile(getClass().getName() + getName(), "acl");
+        acl.deleteOnExit();
+        
+        // Write ACL file
+        PrintWriter aclWriter = new PrintWriter(new FileWriter(acl));
+        for (String line : aclData)
+        {
+            aclWriter.println(line);
+        }
+        aclWriter.close();
+
+        // Load ruleset
+        ConfigurationFile configFile = new PlainConfiguration(acl);
+        configFile.load();
+    }
+
+    public void testMissingACLConfig() throws Exception
+    {
+        try
+        {
+            // Load ruleset
+            ConfigurationFile configFile = new PlainConfiguration(new File("doesnotexist"));
+            configFile.load();
+            
+            fail("fail");
+        }
+        catch (ConfigurationException ce)
+        {
+            assertEquals(String.format(PlainConfiguration.CONFIG_NOT_FOUND_MSG, "doesnotexist"), ce.getMessage());
+            assertTrue(ce.getCause() instanceof FileNotFoundException);
+            assertEquals("doesnotexist (No such file or directory)", ce.getCause().getMessage());
+        }
+    }
+
+    public void testACLFileSyntaxContinuation() throws Exception
+    {
+        try
+        {
+            writeACLConfig("ACL ALLOW ALL \\ ALL");
+            fail("fail");
+        }
+        catch (ConfigurationException ce)
+        {
+            assertEquals(String.format(PlainConfiguration.PREMATURE_CONTINUATION_MSG, 1), ce.getMessage());
+        }
+    }
+
+    public void testACLFileSyntaxTokens() throws Exception
+    {
+        try
+        {
+            writeACLConfig("ACL unparsed ALL ALL");
+            fail("fail");
+        }
+        catch (ConfigurationException ce)
+        {
+            assertEquals(String.format(PlainConfiguration.PARSE_TOKEN_FAILED_MSG, 1), ce.getMessage());
+            assertTrue(ce.getCause() instanceof IllegalArgumentException);
+            assertEquals("Not a valid permission: unparsed", ce.getCause().getMessage());
+        }
+    }
+
+    public void testACLFileSyntaxNotEnoughGroup() throws Exception
+    {
+        try
+        {
+            writeACLConfig("GROUP blah");
+            fail("fail");
+        }
+        catch (ConfigurationException ce)
+        {
+            assertEquals(String.format(PlainConfiguration.NOT_ENOUGH_GROUP_MSG, 1), ce.getMessage());
+        }
+    }
+
+    public void testACLFileSyntaxNotEnoughACL() throws Exception
+    {
+        try
+        {
+            writeACLConfig("ACL ALLOW");
+            fail("fail");
+        }
+        catch (ConfigurationException ce)
+        {
+            assertEquals(String.format(PlainConfiguration.NOT_ENOUGH_ACL_MSG, 1), ce.getMessage());
+        }
+    }
+
+    public void testACLFileSyntaxNotEnoughConfig() throws Exception
+    {
+        try
+        {
+            writeACLConfig("CONFIG");
+            fail("fail");
+        }
+        catch (ConfigurationException ce)
+        {
+            assertEquals(String.format(PlainConfiguration.NOT_ENOUGH_TOKENS_MSG, 1), ce.getMessage());
+        }
+    }
+
+    public void testACLFileSyntaxNotEnough() throws Exception
+    {
+        try
+        {
+            writeACLConfig("INVALID");
+            fail("fail");
+        }
+        catch (ConfigurationException ce)
+        {
+            assertEquals(String.format(PlainConfiguration.NOT_ENOUGH_TOKENS_MSG, 1), ce.getMessage());
+        }
+    }
+
+    public void testACLFileSyntaxPropertyKeyOnly() throws Exception
+    {
+        try
+        {
+            writeACLConfig("ACL ALLOW adk CREATE QUEUE name");
+            fail("fail");
+        }
+        catch (ConfigurationException ce)
+        {
+            assertEquals(String.format(PlainConfiguration.PROPERTY_KEY_ONLY_MSG, 1), ce.getMessage());
+        }
+    }
+
+    public void testACLFileSyntaxPropertyNoEquals() throws Exception
+    {
+        try
+        {
+            writeACLConfig("ACL ALLOW adk CREATE QUEUE name test");
+            fail("fail");
+        }
+        catch (ConfigurationException ce)
+        {
+            assertEquals(String.format(PlainConfiguration.PROPERTY_NO_EQUALS_MSG, 1), ce.getMessage());
+        }
+    }
+
+    public void testACLFileSyntaxPropertyNoValue() throws Exception
+    {
+        try
+        {
+            writeACLConfig("ACL ALLOW adk CREATE QUEUE name =");
+            fail("fail");
+        }
+        catch (ConfigurationException ce)
+        {
+            assertEquals(String.format(PlainConfiguration.PROPERTY_NO_VALUE_MSG, 1), ce.getMessage());
+        }
+    }
+}
diff --git a/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/RuleSetTest.java b/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/RuleSetTest.java
index aad7290557..bd9deac153 100644
--- a/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/RuleSetTest.java
+++ b/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/RuleSetTest.java
@@ -21,13 +21,21 @@
 
 package org.apache.qpid.server.security.access.plugins;
 
+import java.security.Principal;
+import java.util.Arrays;
+
+import javax.security.auth.Subject;
+
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.server.security.Result;
 import org.apache.qpid.server.security.access.ObjectProperties;
 import org.apache.qpid.server.security.access.ObjectType;
 import org.apache.qpid.server.security.access.Operation;
 import org.apache.qpid.server.security.access.Permission;
+import org.apache.qpid.server.security.access.config.Rule;
 import org.apache.qpid.server.security.access.config.RuleSet;
+import org.apache.qpid.server.security.auth.sasl.TestPrincipalUtils;
+import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
 import org.apache.qpid.test.utils.QpidTestCase;
 
 /**
@@ -36,16 +44,24 @@ import org.apache.qpid.test.utils.QpidTestCase;
  * The ruleset is configured directly rather than using an external file by adding rules individually, calling the
  * {@link RuleSet#grant(Integer, String, Permission, Operation, ObjectType, ObjectProperties)} method. Then, the
  * access control mechanism is validated by checking whether operations would be authorised by calling the
- * {@link RuleSet#check(String, Operation, ObjectType, ObjectProperties)} method.
+ * {@link RuleSet#check(Principal, Operation, ObjectType, ObjectProperties)} method.
+ * 
+ * It ensure that permissions can be granted correctly on users directly, ACL groups (that is those
+ * groups declared directly in the ACL itself), and External groups (that is a group from an External
+ * Authentication Provider, such as an LDAP).
+
  */
 public class RuleSetTest extends QpidTestCase
 {
-    private RuleSet _ruleSet;
+    private RuleSet _ruleSet; // Object under test
+
+    private static final String TEST_USER = "user";
 
     // Common things that are passed to frame constructors
     private AMQShortString _queueName = new AMQShortString(this.getClass().getName() + "queue");
     private AMQShortString _exchangeName = new AMQShortString("amq.direct");
     private AMQShortString _exchangeType = new AMQShortString("direct");
+    private Subject _testSubject = TestPrincipalUtils.createTestSubject(TEST_USER);
 
     @Override
     public void setUp() throws Exception
@@ -63,34 +79,36 @@ public class RuleSetTest extends QpidTestCase
         super.tearDown();
     }
 
-    public void assertDenyGrantAllow(String identity, Operation operation, ObjectType objectType)
+    public void assertDenyGrantAllow(Subject subject, Operation operation, ObjectType objectType)
     {
-        assertDenyGrantAllow(identity, operation, objectType, ObjectProperties.EMPTY);
+        assertDenyGrantAllow(subject, operation, objectType, ObjectProperties.EMPTY);
     }
     
-    public void assertDenyGrantAllow(String identity, Operation operation, ObjectType objectType, ObjectProperties properties)
+    public void assertDenyGrantAllow(Subject subject, Operation operation, ObjectType objectType, ObjectProperties properties)
     {
-        assertEquals(Result.DENIED, _ruleSet.check(identity, operation, objectType, properties));
-        _ruleSet.grant(0, identity, Permission.ALLOW, operation, objectType, properties);
+        final Principal identity = UsernamePrincipal.getUsernamePrincipalFromSubject(subject);
+        
+        assertEquals(Result.DENIED, _ruleSet.check(subject, operation, objectType, properties));
+        _ruleSet.grant(0, identity.getName(), Permission.ALLOW, operation, objectType, properties);
         assertEquals(1, _ruleSet.getRuleCount());
-        assertEquals(Result.ALLOWED, _ruleSet.check(identity, operation, objectType, properties));
+        assertEquals(Result.ALLOWED, _ruleSet.check(subject, operation, objectType, properties));
     }
 
     public void testEmptyRuleSet()
     {
         assertNotNull(_ruleSet);
         assertEquals(_ruleSet.getRuleCount(), 0);
-        assertEquals(_ruleSet.getDefault(), _ruleSet.check("user", Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
+        assertEquals(_ruleSet.getDefault(), _ruleSet.check(_testSubject, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
     }
     
     public void testVirtualHostAccess() throws Exception
     {
-        assertDenyGrantAllow("user", Operation.ACCESS, ObjectType.VIRTUALHOST);
+        assertDenyGrantAllow(_testSubject, Operation.ACCESS, ObjectType.VIRTUALHOST);
     }
 
     public void testQueueCreateNamed() throws Exception
     {
-        assertDenyGrantAllow("user", Operation.CREATE, ObjectType.QUEUE, new ObjectProperties(_queueName));
+        assertDenyGrantAllow(_testSubject, Operation.CREATE, ObjectType.QUEUE, new ObjectProperties(_queueName));
     }
 
     public void testQueueCreatenamedNullRoutingKey()
@@ -98,7 +116,7 @@ public class RuleSetTest extends QpidTestCase
         ObjectProperties properties = new ObjectProperties(_queueName);
         properties.put(ObjectProperties.Property.ROUTING_KEY, (String) null);
         
-        assertDenyGrantAllow("user", Operation.CREATE, ObjectType.QUEUE, properties);
+        assertDenyGrantAllow(_testSubject, Operation.CREATE, ObjectType.QUEUE, properties);
     }
 
     public void testExchangeCreate()
@@ -106,17 +124,17 @@ public class RuleSetTest extends QpidTestCase
         ObjectProperties properties = new ObjectProperties(_exchangeName);
         properties.put(ObjectProperties.Property.TYPE, _exchangeType.asString());
         
-        assertDenyGrantAllow("user", Operation.CREATE, ObjectType.EXCHANGE, properties);
+        assertDenyGrantAllow(_testSubject, Operation.CREATE, ObjectType.EXCHANGE, properties);
     }
 
     public void testConsume()
     {
-        assertDenyGrantAllow("user", Operation.CONSUME, ObjectType.QUEUE);
+        assertDenyGrantAllow(_testSubject, Operation.CONSUME, ObjectType.QUEUE);
     }
 
     public void testPublish()
     {
-        assertDenyGrantAllow("user", Operation.PUBLISH, ObjectType.EXCHANGE);
+        assertDenyGrantAllow(_testSubject, Operation.PUBLISH, ObjectType.EXCHANGE);
     }
 
     /**
@@ -131,13 +149,13 @@ public class RuleSetTest extends QpidTestCase
         ObjectProperties normal = new ObjectProperties();
         normal.put(ObjectProperties.Property.AUTO_DELETE, Boolean.FALSE);
         
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CONSUME, ObjectType.QUEUE, temporary));
-        _ruleSet.grant(0, "user", Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, temporary);
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, temporary));
+        _ruleSet.grant(0, TEST_USER, Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, temporary);
         assertEquals(1, _ruleSet.getRuleCount());
-        assertEquals(Result.ALLOWED, _ruleSet.check("user", Operation.CONSUME, ObjectType.QUEUE, temporary));
+        assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, temporary));
         
         // defer to global if exists, otherwise default answer - this is handled by the security manager
-        assertEquals(Result.DEFER, _ruleSet.check("user", Operation.CONSUME, ObjectType.QUEUE, normal));
+        assertEquals(Result.DEFER, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, normal));
     }
 
     /**
@@ -151,15 +169,15 @@ public class RuleSetTest extends QpidTestCase
         ObjectProperties normal = new ObjectProperties(_queueName);
         normal.put(ObjectProperties.Property.AUTO_DELETE, Boolean.FALSE);
         
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CONSUME, ObjectType.QUEUE, temporary));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, temporary));
 
         // should not matter if the temporary permission is processed first or last
-        _ruleSet.grant(1, "user", Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, normal);
-        _ruleSet.grant(2, "user", Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, temporary);
+        _ruleSet.grant(1, TEST_USER, Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, normal);
+        _ruleSet.grant(2, TEST_USER, Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, temporary);
         assertEquals(2, _ruleSet.getRuleCount());
         
-        assertEquals(Result.ALLOWED, _ruleSet.check("user", Operation.CONSUME, ObjectType.QUEUE, normal));
-        assertEquals(Result.ALLOWED, _ruleSet.check("user", Operation.CONSUME, ObjectType.QUEUE, temporary));
+        assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, normal));
+        assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, temporary));
     }
 
     /**
@@ -173,15 +191,15 @@ public class RuleSetTest extends QpidTestCase
         ObjectProperties normal = new ObjectProperties(_queueName);
         normal.put(ObjectProperties.Property.AUTO_DELETE, Boolean.FALSE);
         
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CONSUME, ObjectType.QUEUE, temporary));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, temporary));
 
         // should not matter if the temporary permission is processed first or last
-        _ruleSet.grant(1, "user", Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, temporary);
-        _ruleSet.grant(2, "user", Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, normal);
+        _ruleSet.grant(1, TEST_USER, Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, temporary);
+        _ruleSet.grant(2, TEST_USER, Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, normal);
         assertEquals(2, _ruleSet.getRuleCount());
         
-        assertEquals(Result.ALLOWED, _ruleSet.check("user", Operation.CONSUME, ObjectType.QUEUE, normal));
-        assertEquals(Result.ALLOWED, _ruleSet.check("user", Operation.CONSUME, ObjectType.QUEUE, temporary));
+        assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, normal));
+        assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, temporary));
     }
 
     /*
@@ -197,15 +215,15 @@ public class RuleSetTest extends QpidTestCase
         ObjectProperties namedTemporary = new ObjectProperties(_queueName);
         namedTemporary.put(ObjectProperties.Property.AUTO_DELETE, Boolean.TRUE);
         
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, named));
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, namedTemporary));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
 
-        _ruleSet.grant(1, "user", Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, named);
-        _ruleSet.grant(2, "user", Permission.DENY, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
+        _ruleSet.grant(1, TEST_USER, Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, named);
+        _ruleSet.grant(2, TEST_USER, Permission.DENY, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
         assertEquals(2, _ruleSet.getRuleCount());
         
-        assertEquals(Result.ALLOWED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, named));
-        assertEquals(Result.ALLOWED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, namedTemporary));
+        assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
+        assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
     }
     
     /**
@@ -217,15 +235,15 @@ public class RuleSetTest extends QpidTestCase
         ObjectProperties namedTemporary = new ObjectProperties(_queueName);
         namedTemporary.put(ObjectProperties.Property.AUTO_DELETE, Boolean.TRUE);
         
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, named));
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, namedTemporary));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
 
-        _ruleSet.grant(1, "user", Permission.DENY, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
-        _ruleSet.grant(2, "user", Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, named);
+        _ruleSet.grant(1, TEST_USER, Permission.DENY, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
+        _ruleSet.grant(2, TEST_USER, Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, named);
         assertEquals(2, _ruleSet.getRuleCount());
         
-        assertEquals(Result.ALLOWED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, named));
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, namedTemporary));
+        assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
     }
     
     /**
@@ -239,18 +257,18 @@ public class RuleSetTest extends QpidTestCase
         ObjectProperties namedDurable = new ObjectProperties(_queueName);
         namedDurable.put(ObjectProperties.Property.DURABLE, Boolean.TRUE);
         
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, named));
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, namedTemporary));
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, namedDurable));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedDurable));
 
-        _ruleSet.grant(1, "user", Permission.DENY, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
-        _ruleSet.grant(2, "user", Permission.DENY, Operation.CREATE, ObjectType.QUEUE, namedDurable);
-        _ruleSet.grant(3, "user", Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, named);
+        _ruleSet.grant(1, TEST_USER, Permission.DENY, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
+        _ruleSet.grant(2, TEST_USER, Permission.DENY, Operation.CREATE, ObjectType.QUEUE, namedDurable);
+        _ruleSet.grant(3, TEST_USER, Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, named);
         assertEquals(3, _ruleSet.getRuleCount());
         
-        assertEquals(Result.ALLOWED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, named));
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, namedTemporary));
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, namedDurable));
+        assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedDurable));
     }
     
     public void testNamedTemporaryQueueAllowed()
@@ -259,15 +277,15 @@ public class RuleSetTest extends QpidTestCase
         ObjectProperties namedTemporary = new ObjectProperties(_queueName);
         namedTemporary.put(ObjectProperties.Property.AUTO_DELETE, Boolean.TRUE);
         
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, named));
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, namedTemporary));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
 
-        _ruleSet.grant(1, "user", Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
-        _ruleSet.grant(2, "user", Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, named);
+        _ruleSet.grant(1, TEST_USER, Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
+        _ruleSet.grant(2, TEST_USER, Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, named);
         assertEquals(2, _ruleSet.getRuleCount());
         
-        assertEquals(Result.ALLOWED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, named));
-        assertEquals(Result.ALLOWED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, namedTemporary));
+        assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
+        assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
     }
     
     public void testNamedTemporaryQueueDeniedAllowed()
@@ -276,14 +294,101 @@ public class RuleSetTest extends QpidTestCase
         ObjectProperties namedTemporary = new ObjectProperties(_queueName);
         namedTemporary.put(ObjectProperties.Property.AUTO_DELETE, Boolean.TRUE);
         
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, named));
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, namedTemporary));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
 
-        _ruleSet.grant(1, "user", Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
-        _ruleSet.grant(2, "user", Permission.DENY, Operation.CREATE, ObjectType.QUEUE, named);
+        _ruleSet.grant(1, TEST_USER, Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
+        _ruleSet.grant(2, TEST_USER, Permission.DENY, Operation.CREATE, ObjectType.QUEUE, named);
         assertEquals(2, _ruleSet.getRuleCount());
         
-        assertEquals(Result.DENIED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, named));
-        assertEquals(Result.ALLOWED, _ruleSet.check("user", Operation.CREATE, ObjectType.QUEUE, namedTemporary));
+        assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
+        assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
+    }
+
+    /** 
+     * Tests support for the {@link Rule#ALL} keyword. 
+     */
+    public void testAllowToAll()
+    {
+        _ruleSet.grant(1, Rule.ALL, Permission.ALLOW, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        assertEquals(1, _ruleSet.getRuleCount());
+
+        assertEquals(Result.ALLOWED, _ruleSet.check(TestPrincipalUtils.createTestSubject("usera"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
+        assertEquals(Result.ALLOWED, _ruleSet.check(TestPrincipalUtils.createTestSubject("userb"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
+    }
+
+    /** 
+     * Tests support for ACL groups (i.e. inline groups declared in the ACL file itself).
+     */
+    public void testAclGroupsSupported()
+    {
+        assertTrue(_ruleSet.addGroup("aclgroup", Arrays.asList(new String[] {"usera", "userb"})));  
+        
+        _ruleSet.grant(1, "aclgroup", Permission.ALLOW, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        assertEquals(1, _ruleSet.getRuleCount());
+
+        assertEquals(Result.ALLOWED, _ruleSet.check(TestPrincipalUtils.createTestSubject("usera"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
+        assertEquals(Result.ALLOWED, _ruleSet.check(TestPrincipalUtils.createTestSubject("userb"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
+        assertEquals(Result.DEFER, _ruleSet.check(TestPrincipalUtils.createTestSubject("userc"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
+    }
+
+    /** 
+     * Tests support for nested ACL groups.
+     */
+    public void testNestedAclGroupsSupported()
+    {
+        assertTrue(_ruleSet.addGroup("aclgroup1", Arrays.asList(new String[] {"userb"})));
+        assertTrue(_ruleSet.addGroup("aclgroup2", Arrays.asList(new String[] {"usera", "aclgroup1"})));  
+        
+        _ruleSet.grant(1, "aclgroup2", Permission.ALLOW, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        assertEquals(1, _ruleSet.getRuleCount());
+
+        assertEquals(Result.ALLOWED, _ruleSet.check(TestPrincipalUtils.createTestSubject("usera"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
+        assertEquals(Result.ALLOWED, _ruleSet.check(TestPrincipalUtils.createTestSubject("userb"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
+    }
+
+    /** 
+     * Tests support for nested External groups (i.e. those groups coming from an external source such as an LDAP).
+     */
+    public void testExternalGroupsSupported()
+    {
+        _ruleSet.grant(1, "extgroup1", Permission.ALLOW, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        _ruleSet.grant(2, "extgroup2", Permission.DENY, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        assertEquals(2, _ruleSet.getRuleCount());
+
+        assertEquals(Result.ALLOWED, _ruleSet.check(TestPrincipalUtils.createTestSubject("usera", "extgroup1"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
+        assertEquals(Result.DENIED, _ruleSet.check(TestPrincipalUtils.createTestSubject("userb", "extgroup2"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
+    }
+    
+    /**
+     * Rule order in the ACL determines the outcome of the check.  This test ensures that a user who is
+     * granted explicit permission on an object, is granted that access even although late a group
+     * to which the user belongs is later denied the permission.
+     */
+    public void testAllowDeterminedByRuleOrder()
+    {
+        assertTrue(_ruleSet.addGroup("aclgroup", Arrays.asList(new String[] {"usera"})));
+        
+        _ruleSet.grant(1, "usera", Permission.ALLOW, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        _ruleSet.grant(2, "aclgroup", Permission.DENY, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        assertEquals(2, _ruleSet.getRuleCount());
+
+        assertEquals(Result.ALLOWED, _ruleSet.check(TestPrincipalUtils.createTestSubject("usera"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
+    }
+
+    /**
+     * Rule order in the ACL determines the outcome of the check.  This tests ensures that a user who is denied
+     * access by group, is denied access, despite there being a later rule granting permission to that user.
+     */
+    public void testDenyDeterminedByRuleOrder()
+    {
+        assertTrue(_ruleSet.addGroup("aclgroup", Arrays.asList(new String[] {"usera"})));
+        
+        _ruleSet.grant(1, "aclgroup", Permission.DENY, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        _ruleSet.grant(2, "usera", Permission.ALLOW, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
+        
+        assertEquals(2, _ruleSet.getRuleCount());
+
+        assertEquals(Result.DENIED, _ruleSet.check(TestPrincipalUtils.createTestSubject("usera"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
     }
 }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
index 08eb05680c..8141533045 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
@@ -1083,7 +1083,7 @@ public class AMQChannel implements SessionConfig, AMQSessionModel
                     ? ((BasicContentHeaderProperties) header.getProperties()).getUserId()
                     : null;
 
-        return (!MSG_AUTH || _session.getPrincipal().getName().equals(userID == null? "" : userID.toString()));
+        return (!MSG_AUTH || _session.getAuthorizedPrincipal().getName().equals(userID == null? "" : userID.toString()));
 
     }
 
diff --git a/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java b/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java
index fa2fb9ead1..f21158cd0c 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java
@@ -258,7 +258,6 @@ public class BrokerLink implements LinkConfig, ConnectionListener
                     _remoteFederationTag = UUID.fromString(_transport+":"+_host+":"+_port).toString();
                 }
                 _qpidConnection.setSessionFactory(new SessionFactory());
-                _qpidConnection.setAuthorizationID(_username == null ? "" : _username);
 
                 updateState(State.ESTABLISHING, State.OPERATIONAL);
 
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java
index 79de0678f0..09f35da41d 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java
@@ -23,7 +23,6 @@ package org.apache.qpid.server.handler;
 
 import javax.security.sasl.SaslException;
 import javax.security.sasl.SaslServer;
-
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.ConnectionCloseBody;
@@ -89,7 +88,10 @@ public class ConnectionSecureOkMethodHandler implements StateAwareMethodListener
                 disposeSaslServer(session);
                 break;
             case SUCCESS:
-                _logger.info("Connected as: " + ss.getAuthorizationID());
+                if (_logger.isInfoEnabled())
+                {
+                    _logger.info("Connected as: " + UsernamePrincipal.getUsernamePrincipalFromSubject(authResult.getSubject()));
+                }
                 stateManager.changeState(AMQState.CONNECTION_NOT_TUNED);
 
                 ConnectionTuneBody tuneBody =
@@ -97,8 +99,7 @@ public class ConnectionSecureOkMethodHandler implements StateAwareMethodListener
                                                                 ConnectionStartOkMethodHandler.getConfiguredFrameSize(),
                                                                 ApplicationRegistry.getInstance().getConfiguration().getHeartBeatDelay());
                 session.writeFrame(tuneBody.generateFrame(0));
-                final UsernamePrincipal principal = UsernamePrincipal.getUsernamePrincipalFromSubject(authResult.getSubject());
-                session.setAuthorizedID(principal);
+                session.setAuthorizedSubject(authResult.getSubject());
                 disposeSaslServer(session);                
                 break;
             case CONTINUE:
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
index 544bd62ed8..2dd9a63540 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
@@ -65,7 +65,6 @@ public class ConnectionStartOkMethodHandler implements StateAwareMethodListener<
         _logger.info("Locale selected: " + body.getLocale());
 
         AuthenticationManager authMgr = ApplicationRegistry.getInstance().getAuthenticationManager();
-
         SaslServer ss = null;
         try
         {                       
@@ -78,8 +77,7 @@ public class ConnectionStartOkMethodHandler implements StateAwareMethodListener<
 
             session.setSaslServer(ss);
 
-            AuthenticationResult authResult = authMgr.authenticate(ss, body.getResponse());
-
+            final AuthenticationResult authResult = authMgr.authenticate(ss, body.getResponse());
             //save clientProperties
             if (session.getClientProperties() == null)
             {
@@ -108,8 +106,11 @@ public class ConnectionStartOkMethodHandler implements StateAwareMethodListener<
                     break;
 
                 case SUCCESS:
-                    _logger.info("Connected as: " + ss.getAuthorizationID());
-                    session.setAuthorizedID(new UsernamePrincipal(ss.getAuthorizationID()));
+                    if (_logger.isInfoEnabled())
+                    {
+                        _logger.info("Connected as: " + UsernamePrincipal.getUsernamePrincipalFromSubject(authResult.getSubject()));
+                    }
+                    session.setAuthorizedSubject(authResult.getSubject());
 
                     stateManager.changeState(AMQState.CONNECTION_NOT_TUNED);
 
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
index 8939cfa334..0cfed77f2e 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
@@ -106,7 +106,7 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar
                 else
                 {
                     queue = createQueue(queueName, body, virtualHost, protocolConnection);
-                    queue.setPrincipalHolder(protocolConnection);
+                    queue.setAuthorizationHolder(protocolConnection);
                     if (queue.isDurable() && !queue.isAutoDelete())
                     {
                         store.createQueue(queue, body.getArguments());
@@ -119,7 +119,7 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar
                     if (body.getExclusive())
                     {
                         queue.setExclusiveOwningSession(protocolConnection.getChannel(channelId));
-                        queue.setPrincipalHolder(protocolConnection);
+                        queue.setAuthorizationHolder(protocolConnection);
 
                         if(!body.getDurable())
                         {
diff --git a/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java b/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
index f28873940b..9b357403a8 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
@@ -47,7 +47,7 @@ public class ChannelLogSubject extends AbstractLogSubject
          */
         setLogStringWithFormat(CHANNEL_FORMAT,
                                session.getSessionID(),
-                               session.getPrincipal().getName(),
+                               session.getAuthorizedPrincipal().getName(),
                                session.getRemoteAddress(),
                                session.getVirtualHost().getName(),
                                channel.getChannelId());
diff --git a/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java b/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java
index a697029d24..c1c836f9b4 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java
@@ -56,7 +56,7 @@ public class ConnectionLogSubject extends AbstractLogSubject
     {
         if (!_upToDate)
         {
-            if (_session.getPrincipal() != null)
+            if (_session.getAuthorizedPrincipal() != null)
             {
                 if (_session.getVirtualHost() != null)
                 {
@@ -72,7 +72,7 @@ public class ConnectionLogSubject extends AbstractLogSubject
                      */
                     _logString = "[" + MessageFormat.format(CONNECTION_FORMAT, 
                                                             _session.getSessionID(), 
-                                                            _session.getPrincipal().getName(), 
+                                                            _session.getAuthorizedPrincipal().getName(), 
                                                             _session.getRemoteAddress(),
                                                             _session.getVirtualHost().getName()) 
                                  + "] ";
@@ -83,7 +83,7 @@ public class ConnectionLogSubject extends AbstractLogSubject
                 {
                     _logString = "[" + MessageFormat.format(USER_FORMAT, 
                                                             _session.getSessionID(), 
-                                                            _session.getPrincipal().getName(), 
+                                                            _session.getAuthorizedPrincipal().getName(), 
                                                             _session.getRemoteAddress())
                                  + "] ";
 
diff --git a/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java b/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java
index ce6bd3ee33..68f7689283 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java
@@ -26,7 +26,6 @@ import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
 import java.security.AccessControlContext;
 import java.security.AccessController;
-import java.security.Principal;
 import java.util.Set;
 
 import javax.management.Attribute;
@@ -43,7 +42,6 @@ import javax.management.remote.MBeanServerForwarder;
 import javax.security.auth.Subject;
 
 import org.apache.log4j.Logger;
-import org.apache.qpid.server.logging.actors.CurrentActor;
 import org.apache.qpid.server.logging.actors.ManagementActor;
 import org.apache.qpid.server.logging.messages.ManagementConsoleMessages;
 import org.apache.qpid.server.registry.ApplicationRegistry;
@@ -51,17 +49,13 @@ import org.apache.qpid.server.security.SecurityManager;
 import org.apache.qpid.server.security.access.Operation;
 
 /**
- * This class can be used by the JMXConnectorServer as an InvocationHandler for the mbean operations. This implements
- * the logic for allowing the users to invoke MBean operations and implements the restrictions for readOnly, readWrite
- * and admin users.
+ * This class can be used by the JMXConnectorServer as an InvocationHandler for the mbean operations. It delegates
+ * JMX access decisions to the SecurityPlugin.
  */
 public class MBeanInvocationHandlerImpl implements InvocationHandler, NotificationListener
 {
     private static final Logger _logger = Logger.getLogger(MBeanInvocationHandlerImpl.class);
 
-    public final static String ADMIN = "admin";
-    public final static String READWRITE = "readwrite";
-    public final static String READONLY = "readonly";
     private final static String DELEGATE = "JMImplementation:type=MBeanServerDelegate";
     private MBeanServer _mbs;
     private static ManagementActor  _logActor;
@@ -135,14 +129,13 @@ public class MBeanInvocationHandlerImpl implements InvocationHandler, Notificati
             Set<JMXPrincipal> principals = subject.getPrincipals(JMXPrincipal.class);
             if (principals == null || principals.isEmpty())
             {
-                throw new SecurityException("Access denied: no principal");
+                throw new SecurityException("Access denied: no JMX principal");
             }
-			
-            // Save the principal
-            Principal principal = principals.iterator().next();
-            SecurityManager.setThreadPrincipal(principal);
-    
-			// Get the component, type and impact, which may be null
+
+            // Save the subject
+            SecurityManager.setThreadSubject(subject);
+   
+            // Get the component, type and impact, which may be null
             String type = getType(method, args);
             String vhost = getVirtualHost(method, args);
             int impact = getImpact(method, args);
@@ -284,7 +277,7 @@ public class MBeanInvocationHandlerImpl implements InvocationHandler, Notificati
             }
             catch (JMException ex)
             {
-                ex.printStackTrace();
+                _logger.error("Unable to determine mbean impact for method : " + mbeanMethod, ex);
             }
         }
 
diff --git a/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java b/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java
index 3107185006..4e40305dbb 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java
@@ -167,7 +167,8 @@ public class PluginManager implements Closeable
                 "org.apache.commons.logging; version=1.0.0," +
                 "org.apache.log4j; version=1.2.12," +
                 "javax.management.openmbean; version=1.0.0," +
-                "javax.management; version=1.0.0"
+                "javax.management; version=1.0.0," +
+                "javax.security.auth; version=1.0.0"
             );
         
         // No automatic shutdown hook
diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
index 08f276ae72..d6201f7cf6 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
@@ -37,6 +37,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
 import javax.management.JMException;
+import javax.security.auth.Subject;
 import javax.security.sasl.SaslServer;
 
 import org.apache.log4j.Logger;
@@ -88,6 +89,7 @@ import org.apache.qpid.server.management.ManagedObject;
 import org.apache.qpid.server.output.ProtocolOutputConverter;
 import org.apache.qpid.server.output.ProtocolOutputConverterRegistry;
 import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
 import org.apache.qpid.server.state.AMQState;
 import org.apache.qpid.server.state.AMQStateManager;
 import org.apache.qpid.server.stats.StatisticsCounter;
@@ -145,7 +147,7 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol
 
     private Map<Integer, Long> _closingChannelsList = new ConcurrentHashMap<Integer, Long>();
     private ProtocolOutputConverter _protocolOutputConverter;
-    private Principal _authorizedID;
+    private Subject _authorizedSubject;
     private MethodDispatcher _dispatcher;
     private ProtocolSessionIdentifier _sessionIdentifier;
 
@@ -806,7 +808,7 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol
 
     public String toString()
     {
-        return getRemoteAddress() + "(" + (getAuthorizedID() == null ? "?" : getAuthorizedID().getName() + ")");
+        return getRemoteAddress() + "(" + (getAuthorizedPrincipal() == null ? "?" : getAuthorizedPrincipal().getName() + ")");
     }
 
     public String dump()
@@ -953,19 +955,23 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol
         return _protocolOutputConverter;
     }
 
-    public void setAuthorizedID(Principal authorizedID)
+    public void setAuthorizedSubject(final Subject authorizedSubject)
     {
-        _authorizedID = authorizedID;
+        if (authorizedSubject == null)
+        {
+            throw new IllegalArgumentException("authorizedSubject cannot be null");
+        }
+        _authorizedSubject = authorizedSubject;
     }
-
-    public Principal getAuthorizedID()
+    
+    public Subject getAuthorizedSubject()
     {
-        return _authorizedID;
+        return _authorizedSubject;
     }
-
-    public Principal getPrincipal()
+    
+    public Principal getAuthorizedPrincipal()
     {
-        return _authorizedID;
+        return _authorizedSubject == null ? null : UsernamePrincipal.getUsernamePrincipalFromSubject(_authorizedSubject);
     }
 
     public SocketAddress getRemoteAddress()
@@ -1089,7 +1095,7 @@ public class AMQProtocolEngine implements ProtocolEngine, Managable, AMQProtocol
 
     public String getAuthId()
     {
-        return getAuthorizedID().getName();
+        return getAuthorizedPrincipal().getName();
     }
 
     public Integer getRemotePID()
diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
index c64ed4ad5a..9116bf2767 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
@@ -20,6 +20,7 @@
  */
 package org.apache.qpid.server.protocol;
 
+import javax.security.auth.Subject;
 import javax.security.sasl.SaslServer;
 
 import org.apache.qpid.AMQException;
@@ -28,16 +29,15 @@ import org.apache.qpid.framing.*;
 import org.apache.qpid.AMQConnectionException;
 import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
 import org.apache.qpid.server.AMQChannel;
-import org.apache.qpid.server.security.PrincipalHolder;
+import org.apache.qpid.server.security.AuthorizationHolder;
 import org.apache.qpid.server.logging.LogActor;
 import org.apache.qpid.server.output.ProtocolOutputConverter;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 
-import java.security.Principal;
 import java.util.List;
 
 
-public interface AMQProtocolSession extends AMQVersionAwareProtocolSession, PrincipalHolder, AMQConnectionModel
+public interface AMQProtocolSession extends AMQVersionAwareProtocolSession, AuthorizationHolder, AMQConnectionModel
 {
     long getSessionID();
 
@@ -205,7 +205,7 @@ public interface AMQProtocolSession extends AMQVersionAwareProtocolSession, Prin
 
     public ProtocolOutputConverter getProtocolOutputConverter();
 
-    void setAuthorizedID(Principal authorizedID);
+    void setAuthorizedSubject(Subject authorizedSubject);
 
     public java.net.SocketAddress getRemoteAddress();
 
diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java
index fcac78fafa..16d99de492 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java
@@ -131,7 +131,7 @@ public class AMQProtocolSessionMBean extends AMQManagedObject implements Managed
 
     public String getAuthorizedId()
     {
-        return (_protocolSession.getPrincipal() != null ) ? _protocolSession.getPrincipal().getName() : null;
+        return (_protocolSession.getAuthorizedPrincipal() != null ) ? _protocolSession.getAuthorizedPrincipal().getName() : null;
     }
 
     public String getVersion()
diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
index 42a604e3a5..8c62441d59 100755
--- a/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
@@ -127,7 +127,7 @@ public class ProtocolEngine_0_10  extends InputHandler implements ProtocolEngine
 
     public String getAuthId()
     {
-        return _connection.getAuthorizationID();
+        return _connection.getAuthorizedPrincipal() == null ? null : _connection.getAuthorizedPrincipal().getName();
     }
 
     public String getRemoteProcessName()
diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
index 9b9de8333b..9140a13625 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
@@ -32,7 +32,7 @@ import org.apache.qpid.server.exchange.Exchange;
 import org.apache.qpid.server.exchange.ExchangeReferrer;
 import org.apache.qpid.server.management.Managable;
 import org.apache.qpid.server.management.ManagedObject;
-import org.apache.qpid.server.security.PrincipalHolder;
+import org.apache.qpid.server.security.AuthorizationHolder;
 import org.apache.qpid.server.store.TransactionLogResource;
 import org.apache.qpid.server.subscription.Subscription;
 import org.apache.qpid.server.txn.ServerTransaction;
@@ -69,8 +69,8 @@ public interface AMQQueue extends Managable, Comparable<AMQQueue>, ExchangeRefer
     boolean isAutoDelete();
 
     AMQShortString getOwner();
-    PrincipalHolder getPrincipalHolder();
-    void setPrincipalHolder(PrincipalHolder principalHolder);
+    AuthorizationHolder getAuthorizationHolder();
+    void setAuthorizationHolder(AuthorizationHolder principalHolder);
 
     void setExclusiveOwningSession(AMQSessionModel owner);
     AMQSessionModel getExclusiveOwningSession();
diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
index 274cb6714a..4dbe94e4d7 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
@@ -44,7 +44,7 @@ import org.apache.qpid.server.logging.subjects.QueueLogSubject;
 import org.apache.qpid.server.management.ManagedObject;
 import org.apache.qpid.server.message.ServerMessage;
 import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.security.PrincipalHolder;
+import org.apache.qpid.server.security.AuthorizationHolder;
 import org.apache.qpid.server.subscription.Subscription;
 import org.apache.qpid.server.subscription.SubscriptionList;
 import org.apache.qpid.server.txn.AutoCommitTransaction;
@@ -83,7 +83,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
     /** null means shared */
     private final AMQShortString _owner;
 
-    private PrincipalHolder _prinicpalHolder;
+    private AuthorizationHolder _authorizationHolder;
 
     private boolean _exclusive = false;
     private AMQSessionModel _exclusiveOwner;
@@ -373,14 +373,14 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
         return _owner;
     }
 
-    public PrincipalHolder getPrincipalHolder()
+    public AuthorizationHolder getAuthorizationHolder()
     {
-        return _prinicpalHolder;
+        return _authorizationHolder;
     }
 
-    public void setPrincipalHolder(PrincipalHolder prinicpalHolder)
+    public void setAuthorizationHolder(final AuthorizationHolder authorizationHolder)
     {
-        _prinicpalHolder = prinicpalHolder;
+        _authorizationHolder = authorizationHolder;
     }
 
 
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/AuthorizationHolder.java b/java/broker/src/main/java/org/apache/qpid/server/security/AuthorizationHolder.java
new file mode 100755
index 0000000000..3d8c77a86f
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/AuthorizationHolder.java
@@ -0,0 +1,53 @@
+/*
+ *
+ * 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.security;
+
+import java.security.Principal;
+
+import javax.security.auth.Subject;
+
+import org.apache.qpid.server.security.auth.sasl.GroupPrincipal;
+import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
+
+/**
+ * Represents the authorization of the logged on user. 
+ * 
+ */
+public interface AuthorizationHolder
+{
+    /** 
+     * Returns the {@link Subject} of the authorized user.  This is guaranteed to
+     * contain at least one {@link UsernamePrincipal}, representing the the identity
+     * used when the user logged on to the application, and zero or more {@link GroupPrincipal}
+     * representing the group(s) to which the user belongs.
+     *
+     * @return the Subject
+     */
+    Subject getAuthorizedSubject();
+
+    /** 
+     * Returns the {@link Principal} representing the the identity
+     * used when the user logged on to the application.
+     * 
+     * @return a Principal
+     */
+    Principal getAuthorizedPrincipal();
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/SecurityManager.java b/java/broker/src/main/java/org/apache/qpid/server/security/SecurityManager.java
index f18c327692..f582fed6a0 100755
--- a/java/broker/src/main/java/org/apache/qpid/server/security/SecurityManager.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/SecurityManager.java
@@ -18,8 +18,19 @@
  */
 package org.apache.qpid.server.security;
 
-import static org.apache.qpid.server.security.access.ObjectType.*;
-import static org.apache.qpid.server.security.access.Operation.*;
+import static org.apache.qpid.server.security.access.ObjectType.EXCHANGE;
+import static org.apache.qpid.server.security.access.ObjectType.METHOD;
+import static org.apache.qpid.server.security.access.ObjectType.OBJECT;
+import static org.apache.qpid.server.security.access.ObjectType.QUEUE;
+import static org.apache.qpid.server.security.access.ObjectType.VIRTUALHOST;
+import static org.apache.qpid.server.security.access.Operation.ACCESS;
+import static org.apache.qpid.server.security.access.Operation.BIND;
+import static org.apache.qpid.server.security.access.Operation.CONSUME;
+import static org.apache.qpid.server.security.access.Operation.CREATE;
+import static org.apache.qpid.server.security.access.Operation.DELETE;
+import static org.apache.qpid.server.security.access.Operation.PUBLISH;
+import static org.apache.qpid.server.security.access.Operation.PURGE;
+import static org.apache.qpid.server.security.access.Operation.UNBIND;
 
 import java.net.SocketAddress;
 import java.security.Principal;
@@ -29,6 +40,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 
+import javax.security.auth.Subject;
+
 import org.apache.commons.configuration.Configuration;
 import org.apache.commons.configuration.ConfigurationException;
 import org.apache.log4j.Logger;
@@ -37,11 +50,9 @@ import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
 import org.apache.qpid.server.configuration.plugins.ConfigurationPluginFactory;
 import org.apache.qpid.server.exchange.Exchange;
 import org.apache.qpid.server.plugins.PluginManager;
-import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.security.access.ObjectProperties;
 import org.apache.qpid.server.security.access.Operation;
-import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
 
 /**
  * The security manager contains references to all loaded {@link SecurityPlugin}s and delegates security decisions to them based
@@ -55,7 +66,7 @@ public class SecurityManager
     private static final Logger _logger = Logger.getLogger(SecurityManager.class);
     
     /** Container for the {@link Principal} that is using to this thread. */
-    private static final ThreadLocal<Principal> _principal = new ThreadLocal<Principal>();
+    private static final ThreadLocal<Subject> _subject = new ThreadLocal<Subject>();
     
     private PluginManager _pluginManager;
     private Map<String, SecurityPluginFactory> _pluginFactories = new HashMap<String, SecurityPluginFactory>();
@@ -126,19 +137,14 @@ public class SecurityManager
         configureHostPlugins(configuration);
     }
 
-    public static Principal getThreadPrincipal()
-    {
-        return _principal.get();
-    }
-
-    public static void setThreadPrincipal(Principal principal)
+    public static Subject getThreadSubject()
     {
-        _principal.set(principal);
+        return _subject.get();
     }
 
-    public static void setThreadPrincipal(String authId)
+    public static void setThreadSubject(final Subject subject)
     {
-        setThreadPrincipal(new UsernamePrincipal(authId));
+        _subject.set(subject);
     }
 
     public void configureHostPlugins(ConfigurationPlugin hostConfig) throws ConfigurationException
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java
index 70a9ea5356..e4bf8df340 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java
@@ -149,9 +149,9 @@ public class ObjectProperties extends HashMap<ObjectProperties.Property, String>
         {
             put(Property.OWNER, queue.getOwner());
         }
-        else if (queue.getPrincipalHolder() != null)
+        else if (queue.getAuthorizationHolder() != null)
         {
-            put(Property.OWNER, queue.getPrincipalHolder().getPrincipal().getName());
+            put(Property.OWNER, queue.getAuthorizationHolder().getAuthorizedPrincipal().getName());
         }
     }
     
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/GroupPrincipal.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/GroupPrincipal.java
new file mode 100644
index 0000000000..30a503c769
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/GroupPrincipal.java
@@ -0,0 +1,99 @@
+/*
+ *
+ * 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.security.auth.sasl;
+
+import java.security.Principal;
+import java.security.acl.Group;
+import java.util.Enumeration;
+
+/**
+ * Immutable representation of a user group.  In Qpid, groups do <b>not</b> know
+ * about their membership, and therefore the {@link #addMember(Principal)}
+ * methods etc throw {@link UnsupportedOperationException}.
+ *
+ */
+public class GroupPrincipal implements Group
+{
+    /** Name of the group */
+    private final String _groupName;
+    
+    public GroupPrincipal(final String groupName)
+    {
+        _groupName = groupName;
+    }
+
+    public String getName()
+    {
+        return _groupName;
+    }
+
+    public boolean addMember(Principal user)
+    {
+        throw new UnsupportedOperationException("Not supported");
+    }
+
+    public boolean removeMember(Principal user)
+    {
+        throw new UnsupportedOperationException("Not supported");
+    }
+
+    public boolean isMember(Principal member)
+    {
+        throw new UnsupportedOperationException("Not supported");
+    }
+
+    public Enumeration<? extends Principal> members()
+    {
+        throw new UnsupportedOperationException("Not supported");
+    }
+
+    /**
+     * @see java.lang.Object#hashCode()
+     */
+    public int hashCode()
+    {
+        final int prime = 37;
+        return prime * _groupName.hashCode();
+    }
+
+    /**
+     * @see java.lang.Object#equals(java.lang.Object)
+     */
+    public boolean equals(Object obj)
+    {
+        if (this == obj)
+        {
+            return true;
+        }
+        else 
+        {
+            if (obj instanceof GroupPrincipal)
+            {
+                GroupPrincipal other = (GroupPrincipal) obj;
+                return _groupName.equals(other._groupName);
+            }
+            else
+            {
+                return false;
+            }
+        }
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java b/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java
index 6cc5e7b019..33aebffcfb 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java
@@ -259,7 +259,7 @@ public class AMQStateManager implements AMQMethodListener
 
     public AMQProtocolSession getProtocolSession()
     {
-        SecurityManager.setThreadPrincipal(_protocolSession.getPrincipal());
+        SecurityManager.setThreadSubject(_protocolSession.getAuthorizedSubject());
         return _protocolSession;
     }
 }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
index 2113494cc5..9b3673e8b7 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
@@ -22,12 +22,15 @@ package org.apache.qpid.server.transport;
 
 import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.*;
 
+import java.security.Principal;
 import java.text.MessageFormat;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
 
+import javax.security.auth.Subject;
+
 import org.apache.qpid.AMQException;
 import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.server.configuration.ConnectionConfig;
@@ -38,7 +41,8 @@ import org.apache.qpid.server.logging.actors.GenericActor;
 import org.apache.qpid.server.logging.messages.ConnectionMessages;
 import org.apache.qpid.server.protocol.AMQConnectionModel;
 import org.apache.qpid.server.protocol.AMQSessionModel;
-import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.security.AuthorizationHolder;
+import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
 import org.apache.qpid.server.stats.StatisticsCounter;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.transport.Connection;
@@ -49,14 +53,15 @@ import org.apache.qpid.transport.Method;
 import org.apache.qpid.transport.ProtocolEvent;
 import org.apache.qpid.transport.Session;
 
-public class ServerConnection extends Connection implements AMQConnectionModel, LogSubject
+public class ServerConnection extends Connection implements AMQConnectionModel, LogSubject, AuthorizationHolder
 {
     private ConnectionConfig _config;
     private Runnable _onOpenTask;
     private AtomicBoolean _logClosed = new AtomicBoolean(false);
     private LogActor _actor = GenericActor.getInstance(this);
 
-    private ApplicationRegistry _registry;
+    private Subject _authorizedSubject = null;
+    private Principal _authorizedPrincipal = null;
     private boolean _statisticsEnabled = false;
     private StatisticsCounter _messagesDelivered, _dataDelivered, _messagesReceived, _dataReceived;
     
@@ -212,9 +217,9 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
     public String toLogString()
     {
         boolean hasVirtualHost = (null != this.getVirtualHost());
-        boolean hasPrincipal = (null != getAuthorizationID());
+        boolean hasClientId = (null != getClientId());
 
-        if (hasPrincipal && hasVirtualHost)
+        if (hasClientId && hasVirtualHost)
         {
             return "[" +
                     MessageFormat.format(CONNECTION_FORMAT,
@@ -224,7 +229,7 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
                                          getVirtualHost().getName())
                  + "] ";
         }
-        else if (hasPrincipal)
+        else if (hasClientId)
         {
             return "[" +
                     MessageFormat.format(USER_FORMAT,
@@ -341,4 +346,37 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
     {
         _statisticsEnabled = enabled;
     }
+
+    /**
+     * @return authorizedSubject
+     */
+    public Subject getAuthorizedSubject()
+    {
+        return _authorizedSubject;
+    }
+
+    /**
+     * Sets the authorized subject.  It also extracts the UsernamePrincipal from the subject
+     * and caches it for optimisation purposes.
+     *
+     * @param authorizedSubject
+     */
+    public void setAuthorizedSubject(final Subject authorizedSubject)
+    {
+        if (authorizedSubject == null)
+        {
+            _authorizedSubject = null;
+            _authorizedPrincipal = null;
+        }
+        else
+        {
+            _authorizedSubject = authorizedSubject;
+            _authorizedPrincipal = UsernamePrincipal.getUsernamePrincipalFromSubject(_authorizedSubject);
+        }
+    }
+
+    public Principal getAuthorizedPrincipal()
+    {
+        return _authorizedPrincipal;
+    }
 }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
index 174dcbfa69..27e199291d 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
@@ -34,6 +34,8 @@ import org.apache.qpid.protocol.ProtocolEngine;
 import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.registry.IApplicationRegistry;
 import org.apache.qpid.server.security.SecurityManager;
+import org.apache.qpid.server.security.auth.AuthenticationResult;
+import org.apache.qpid.server.security.auth.AuthenticationResult.AuthenticationStatus;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.transport.*;
 
@@ -70,7 +72,6 @@ public class ServerConnectionDelegate extends ServerDelegate
         return list;
     }
 
-    @Override
     public ServerSession getSession(Connection conn, SessionAttach atc)
     {
         SessionDelegate serverSessionDelegate = new ServerSessionDelegate(_appRegistry);
@@ -80,14 +81,33 @@ public class ServerConnectionDelegate extends ServerDelegate
         return ssn;
     }
 
-    @Override
     protected SaslServer createSaslServer(String mechanism) throws SaslException
     {
         return _appRegistry.getAuthenticationManager().createSaslServer(mechanism, _localFQDN);
 
     }
 
-    @Override
+    protected void secure(final SaslServer ss, final Connection conn, final byte[] response)
+    {
+        final AuthenticationResult authResult = _appRegistry.getAuthenticationManager().authenticate(ss, response);
+        final ServerConnection sconn = (ServerConnection) conn;
+        
+        
+        if (AuthenticationStatus.SUCCESS.equals(authResult.getStatus()))
+        {
+            tuneAuthorizedConnection(sconn);
+            sconn.setAuthorizedSubject(authResult.getSubject());
+        }
+        else if (AuthenticationStatus.CONTINUE.equals(authResult.getStatus()))
+        {
+            connectionAuthContinue(sconn, authResult.getChallenge());
+        }
+        else
+        {
+            connectionAuthFailed(sconn, authResult.getCause());
+        }
+    }
+
     public void connectionClose(Connection conn, ConnectionClose close)
     {
         try
@@ -101,10 +121,9 @@ public class ServerConnectionDelegate extends ServerDelegate
         
     }
 
-    @Override
     public void connectionOpen(Connection conn, ConnectionOpen open)
     {
-        ServerConnection sconn = (ServerConnection) conn;
+        final ServerConnection sconn = (ServerConnection) conn;
         
         VirtualHost vhost;
         String vhostName;
@@ -118,7 +137,7 @@ public class ServerConnectionDelegate extends ServerDelegate
         }
         vhost = _appRegistry.getVirtualHostRegistry().getVirtualHost(vhostName);
 
-        SecurityManager.setThreadPrincipal(conn.getAuthorizationID());
+        SecurityManager.setThreadSubject(sconn.getAuthorizedSubject());
         
         if(vhost != null)
         {
diff --git a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
index 5c6206c53c..e9168f71fb 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
@@ -20,8 +20,8 @@
  */
 package org.apache.qpid.server.transport;
 
-import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.*;
-import static org.apache.qpid.util.Serial.*;
+import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CHANNEL_FORMAT;
+import static org.apache.qpid.util.Serial.gt;
 
 import java.lang.ref.WeakReference;
 import java.security.Principal;
@@ -38,6 +38,8 @@ import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicLong;
 
+import javax.security.auth.Subject;
+
 import org.apache.qpid.AMQException;
 import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.protocol.ProtocolEngine;
@@ -57,7 +59,7 @@ import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.queue.BaseQueue;
 import org.apache.qpid.server.queue.QueueEntry;
-import org.apache.qpid.server.security.PrincipalHolder;
+import org.apache.qpid.server.security.AuthorizationHolder;
 import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.subscription.Subscription_0_10;
 import org.apache.qpid.server.txn.AutoCommitTransaction;
@@ -75,9 +77,7 @@ import org.apache.qpid.transport.SessionDelegate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.sun.security.auth.UserPrincipal;
-
-public class ServerSession extends Session implements PrincipalHolder, SessionConfig, AMQSessionModel, LogSubject
+public class ServerSession extends Session implements AuthorizationHolder, SessionConfig, AMQSessionModel, LogSubject
 {
     private static final Logger _logger = LoggerFactory.getLogger(ServerSession.class);
     
@@ -118,8 +118,6 @@ public class ServerSession extends Session implements PrincipalHolder, SessionCo
     private final AtomicLong _txnCount = new AtomicLong(0);
     private final AtomicLong _txnUpdateTime = new AtomicLong(0);
 
-    private Principal _principal;
-
     private Map<String, Subscription_0_10> _subscriptions = new ConcurrentHashMap<String, Subscription_0_10>();
 
     private final List<Task> _taskList = new CopyOnWriteArrayList<Task>();
@@ -146,7 +144,7 @@ public class ServerSession extends Session implements PrincipalHolder, SessionCo
         super(connection, delegate, name, expiry);
         _connectionConfig = connConfig;        
         _transaction = new AutoCommitTransaction(this.getMessageStore());
-        _principal = new UserPrincipal(connection.getAuthorizationID());
+
         _reference = new WeakReference<Session>(this);
         _id = getConfigStore().createId();
         getConfigStore().addConfiguredObject(this);
@@ -419,7 +417,7 @@ public class ServerSession extends Session implements PrincipalHolder, SessionCo
         catch (AMQException e)
         {
             // TODO
-            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
+            _logger.error("Failed to unregister subscription", e);
         }
         finally
         {
@@ -516,9 +514,14 @@ public class ServerSession extends Session implements PrincipalHolder, SessionCo
         return _txnCount.get();
     }
     
-    public Principal getPrincipal()
+    public Principal getAuthorizedPrincipal()
+    {
+        return ((ServerConnection) getConnection()).getAuthorizedPrincipal();
+    }
+    
+    public Subject getAuthorizedSubject()
     {
-        return _principal;
+        return ((ServerConnection) getConnection()).getAuthorizedSubject();
     }
 
     public void addSessionCloseTask(Task task)
diff --git a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
index be659c87ae..4b8b13fc7f 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
@@ -27,19 +27,20 @@ import java.util.Map;
 
 import org.apache.qpid.AMQException;
 import org.apache.qpid.AMQUnknownExchangeType;
-import org.apache.qpid.common.AMQPFilterTypes;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.server.exchange.*;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.exchange.ExchangeFactory;
+import org.apache.qpid.server.exchange.ExchangeInUseException;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.exchange.ExchangeType;
+import org.apache.qpid.server.exchange.HeadersExchange;
 import org.apache.qpid.server.filter.FilterManager;
 import org.apache.qpid.server.filter.FilterManagerFactory;
 import org.apache.qpid.server.flow.FlowCreditManager_0_10;
 import org.apache.qpid.server.flow.WindowCreditManager;
-import org.apache.qpid.server.logging.actors.CurrentActor;
-import org.apache.qpid.server.logging.actors.GenericActor;
 import org.apache.qpid.server.message.MessageMetaData_0_10;
 import org.apache.qpid.server.message.MessageTransferMessage;
-import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.queue.AMQQueueFactory;
 import org.apache.qpid.server.queue.BaseQueue;
@@ -105,7 +106,8 @@ public class ServerSessionDelegate extends SessionDelegate
     @Override
     public void command(Session session, Method method)
     {
-        SecurityManager.setThreadPrincipal(session.getConnection().getAuthorizationID());
+        final ServerConnection scon = (ServerConnection) session.getConnection();
+        SecurityManager.setThreadSubject(scon.getAuthorizedSubject());
 
         if(!session.isClosing())
         {
@@ -203,7 +205,7 @@ public class ServerSessionDelegate extends SessionDelegate
                 {
                     exception(session,method,ExecutionErrorCode.NOT_FOUND, "Queue: " + queueName + " not found");
                 }
-                else if(queue.getPrincipalHolder() != null && queue.getPrincipalHolder() != session)
+                else if(queue.getAuthorizationHolder() != null && queue.getAuthorizationHolder() != session)
                 {
                     exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
                 }
@@ -213,17 +215,17 @@ public class ServerSessionDelegate extends SessionDelegate
                     {
                         ServerSession s = (ServerSession) session;
                         queue.setExclusiveOwningSession(s);
-                        if(queue.getPrincipalHolder() == null)
+                        if(queue.getAuthorizationHolder() == null)
                         {
-                            queue.setPrincipalHolder(s);
+                            queue.setAuthorizationHolder(s);
                             queue.setExclusiveOwningSession(s);
                             ((ServerSession) session).addSessionCloseTask(new ServerSession.Task()
                             {
                                 public void doTask(ServerSession session)
                                 {
-                                    if(queue.getPrincipalHolder() == session)
+                                    if(queue.getAuthorizationHolder() == session)
                                     {
-                                        queue.setPrincipalHolder(null);
+                                        queue.setAuthorizationHolder(null);
                                         queue.setExclusiveOwningSession(null);
                                     }
                                 }
@@ -389,7 +391,7 @@ public class ServerSessionDelegate extends SessionDelegate
             ((ServerSession)session).unregister(sub);
             if(!queue.isDeleted() && queue.isExclusive() && queue.getConsumerCount() == 0)
             {
-                queue.setPrincipalHolder(null);
+                queue.setAuthorizationHolder(null);
             }
         }
     }
@@ -1007,7 +1009,7 @@ public class ServerSessionDelegate extends SessionDelegate
                             {
                                 public void doTask(ServerSession session)
                                 {
-                                    q.setPrincipalHolder(null);
+                                    q.setAuthorizationHolder(null);
                                     q.setExclusiveOwningSession(null);
                                 }
                             };
@@ -1077,7 +1079,7 @@ public class ServerSessionDelegate extends SessionDelegate
             }
             else
             {
-                if(queue.getPrincipalHolder() != null && queue.getPrincipalHolder() != session)
+                if(queue.getAuthorizationHolder() != null && queue.getAuthorizationHolder() != session)
                 {
                     exception(session,method,ExecutionErrorCode.RESOURCE_LOCKED, "Exclusive Queue: " + queueName + " owned exclusively by another session");
                 }
diff --git a/java/broker/src/test/java/org/apache/qpid/server/logging/UnitTestMessageLogger.java b/java/broker/src/test/java/org/apache/qpid/server/logging/UnitTestMessageLogger.java
index 3752dcb37e..e8defd0e58 100644
--- a/java/broker/src/test/java/org/apache/qpid/server/logging/UnitTestMessageLogger.java
+++ b/java/broker/src/test/java/org/apache/qpid/server/logging/UnitTestMessageLogger.java
@@ -28,11 +28,7 @@ import org.apache.qpid.server.logging.AbstractRootMessageLogger;
 
 public class UnitTestMessageLogger extends AbstractRootMessageLogger
 {
-    List<Object> _log;
-    
-    {
-        _log = new LinkedList<Object>();
-    }
+    private final List<Object> _log = new LinkedList<Object>();
     
     public UnitTestMessageLogger()
     {
@@ -69,4 +65,14 @@ public class UnitTestMessageLogger extends AbstractRootMessageLogger
     {
         _log.clear();
     }
+    
+    public boolean messageContains(final int index, final String contains)
+    {
+        if (index + 1 > _log.size())
+        {
+            throw new IllegalArgumentException("Message with index " + index + " has not been logged");
+        }
+        final String message = _log.get(index).toString();
+        return message.contains(contains);
+    }
 }
diff --git a/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java b/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java
index 2b724af2b1..3af665141c 100644
--- a/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java
+++ b/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java
@@ -22,12 +22,15 @@ package org.apache.qpid.server.protocol;
 
 import java.security.Principal;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import javax.security.auth.Subject;
+
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.ContentHeaderBody;
@@ -39,6 +42,8 @@ import org.apache.qpid.server.message.MessageContentSource;
 import org.apache.qpid.server.output.ProtocolOutputConverter;
 import org.apache.qpid.server.queue.QueueEntry;
 import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
+import org.apache.qpid.server.state.AMQState;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.transport.TestNetworkConnection;
 
@@ -55,13 +60,8 @@ public class InternalTestProtocolSession extends AMQProtocolEngine implements Pr
         _channelDelivers = new HashMap<Integer, Map<AMQShortString, LinkedList<DeliveryPair>>>();
 
         // Need to authenticate session for it to be representative testing.
-        setAuthorizedID(new Principal()
-        {
-            public String getName()
-            {
-                return "InternalTestProtocolSession";
-            }
-        });
+        setAuthorizedSubject(new Subject(true, Collections.singleton(new UsernamePrincipal("InternalTestProtocolSession")),
+                Collections.EMPTY_SET, Collections.EMPTY_SET));
 
         setVirtualHost(virtualHost);
     }
diff --git a/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java b/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
index 888a16053c..4c31092983 100644
--- a/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
+++ b/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
@@ -29,7 +29,7 @@ import org.apache.qpid.server.subscription.Subscription;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.server.management.ManagedObject;
 import org.apache.qpid.server.message.ServerMessage;
-import org.apache.qpid.server.security.PrincipalHolder;
+import org.apache.qpid.server.security.AuthorizationHolder;
 import org.apache.qpid.server.AMQChannel;
 import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.binding.Binding;
@@ -48,7 +48,7 @@ public class MockAMQQueue implements AMQQueue
     private AMQShortString _name;
     private VirtualHost _virtualhost;
 
-    private PrincipalHolder _principalHolder;
+    private AuthorizationHolder _authorizationHolder;
 
     private AMQSessionModel _exclusiveOwner;
     private AMQShortString _owner;
@@ -536,14 +536,14 @@ public class MockAMQQueue implements AMQQueue
         return null;  //To change body of implemented methods use File | Settings | File Templates.
     }
 
-    public PrincipalHolder getPrincipalHolder()
+    public AuthorizationHolder getAuthorizationHolder()
     {
-        return _principalHolder;
+        return _authorizationHolder;
     }
 
-    public void setPrincipalHolder(PrincipalHolder principalHolder)
+    public void setAuthorizationHolder(final AuthorizationHolder authorizationHolder)
     {
-        _principalHolder = principalHolder;
+        _authorizationHolder = authorizationHolder;
     }
 
     public AMQSessionModel getExclusiveOwningSession()
diff --git a/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/GroupPrincipalTest.java b/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/GroupPrincipalTest.java
new file mode 100644
index 0000000000..076b7c9248
--- /dev/null
+++ b/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/GroupPrincipalTest.java
@@ -0,0 +1,86 @@
+/*
+ *
+ * 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.security.auth.sasl;
+
+import junit.framework.TestCase;
+
+public class GroupPrincipalTest extends TestCase
+{
+    public void testGetName()
+    {
+        final GroupPrincipal principal = new GroupPrincipal("group");
+        assertEquals("group", principal.getName());
+    }
+
+    public void testAddRejected()
+    {
+        final GroupPrincipal principal = new GroupPrincipal("group");
+        final UsernamePrincipal user = new UsernamePrincipal("name");
+        
+        try
+        {
+            principal.addMember(user);
+            fail("Exception not thrown");
+        }
+        catch (UnsupportedOperationException uso)
+        {
+            // PASS
+        }
+    }
+    
+    public void testEqualitySameName()
+    {
+        final String string = "string";
+        final GroupPrincipal principal1 = new GroupPrincipal(string);
+        final GroupPrincipal principal2 = new GroupPrincipal(string);
+        assertTrue(principal1.equals(principal2));
+    }
+
+    public void testEqualityEqualName()
+    {
+        final GroupPrincipal principal1 = new GroupPrincipal(new String("string"));
+        final GroupPrincipal principal2 = new GroupPrincipal(new String("string"));
+        assertTrue(principal1.equals(principal2));
+    }
+
+    public void testInequalityDifferentGroupPrincipals()
+    {
+        GroupPrincipal principal1 = new GroupPrincipal("string1");
+        GroupPrincipal principal2 = new GroupPrincipal("string2");
+        assertFalse(principal1.equals(principal2));
+    }
+
+    public void testInequalityNonGroupPrincipal()
+    {
+        GroupPrincipal principal = new GroupPrincipal("string");
+        assertFalse(principal.equals(new UsernamePrincipal("string")));
+    }
+
+    public void testInequalityNull()
+    {
+        GroupPrincipal principal = new GroupPrincipal("string");
+        assertFalse(principal.equals(null));
+    }
+
+    
+
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/PrincipalHolder.java b/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/TestPrincipalUtils.java
index 7e93623cab..8b9b2df5a3 100755..100644
--- a/java/broker/src/main/java/org/apache/qpid/server/security/PrincipalHolder.java
+++ b/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/TestPrincipalUtils.java
@@ -18,12 +18,32 @@
  * under the License.
  *
  */
-package org.apache.qpid.server.security;
+package org.apache.qpid.server.security.auth.sasl;
 
 import java.security.Principal;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
 
-public interface PrincipalHolder
+import javax.security.auth.Subject;
+
+public class TestPrincipalUtils
 {
-    /** @return a Principal that was used to authorized this session */
-    Principal getPrincipal();
+
+    /**
+     * Creates a test subject, with exactly one UsernamePrincipal and zero or more GroupPrincipals.
+     */
+    public static Subject createTestSubject(final String username, final String... groups)
+    {
+        final Set<Principal> principals = new HashSet<Principal>(1 + groups.length);
+        principals.add(new UsernamePrincipal(username));
+        for (String group : groups)
+        {
+            principals.add(new GroupPrincipal(group));
+        }
+        
+        final Subject subject = new Subject(true, principals, Collections.EMPTY_SET, Collections.EMPTY_SET);
+        return subject;
+    }
+
 }
diff --git a/java/common/src/main/java/org/apache/qpid/transport/Connection.java b/java/common/src/main/java/org/apache/qpid/transport/Connection.java
index 609611e3fb..f4e3a10f92 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/Connection.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/Connection.java
@@ -120,7 +120,6 @@ public class Connection extends ConnectionInvoker
     private SaslServer saslServer;
     private SaslClient saslClient;
     private int idleTimeout = 0;
-    private String _authorizationID;
     private Map<String,Object> _serverProperties;
     private String userID;
     private ConnectionSettings conSettings;
@@ -661,16 +660,6 @@ public class Connection extends ConnectionInvoker
         return idleTimeout;
     }
 
-    public void setAuthorizationID(String authorizationID)
-    {
-        _authorizationID = authorizationID;
-    }
-
-    public String getAuthorizationID()
-    {
-        return _authorizationID;
-    }
-
     public String getUserID()
     {
         return userID;
diff --git a/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java b/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java
index f21df251da..11af86f412 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java
@@ -75,10 +75,7 @@ public class ServerDelegate extends ConnectionDelegate
 
         if (mechanism == null || mechanism.length() == 0)
         {
-            conn.connectionTune
-                (getChannelMax(),
-                 org.apache.qpid.transport.network.ConnectionBinding.MAX_FRAME_SIZE,
-                 0, getHeartbeatMax());
+            tuneAuthorizedConnection(conn);
             return;
         }
 
@@ -97,8 +94,7 @@ public class ServerDelegate extends ConnectionDelegate
         }
         catch (SaslException e)
         {
-            conn.exception(e);
-            conn.connectionClose(ConnectionCloseCode.CONNECTION_FORCED, e.getMessage());
+            connectionAuthFailed(conn, e);
         }
     }
 
@@ -109,33 +105,52 @@ public class ServerDelegate extends ConnectionDelegate
         return ss;
     }
 
-    private void secure(Connection conn, byte[] response)
+    protected void secure(final SaslServer ss, final Connection conn, final byte[] response)
     {
-        SaslServer ss = conn.getSaslServer();
         try
         {
             byte[] challenge = ss.evaluateResponse(response);
             if (ss.isComplete())
             {
                 ss.dispose();
-                conn.connectionTune
-                    (getChannelMax(),
-                     org.apache.qpid.transport.network.ConnectionBinding.MAX_FRAME_SIZE,
-                     0, getHeartbeatMax());
-                conn.setAuthorizationID(ss.getAuthorizationID());
+                tuneAuthorizedConnection(conn);
             }
             else
             {
-                conn.connectionSecure(challenge);
+                connectionAuthContinue(conn, challenge);
             }
         }
         catch (SaslException e)
         {
-            conn.exception(e);
-            conn.connectionClose(ConnectionCloseCode.CONNECTION_FORCED, e.getMessage());
+            connectionAuthFailed(conn, e);
         }
     }
 
+    protected void connectionAuthFailed(final Connection conn, Exception e)
+    {
+        conn.exception(e);
+        conn.connectionClose(ConnectionCloseCode.CONNECTION_FORCED, e.getMessage());
+    }
+
+    protected void connectionAuthContinue(final Connection conn, byte[] challenge)
+    {
+        conn.connectionSecure(challenge);
+    }
+
+    protected void tuneAuthorizedConnection(final Connection conn)
+    {
+        conn.connectionTune
+            (getChannelMax(),
+             org.apache.qpid.transport.network.ConnectionBinding.MAX_FRAME_SIZE,
+             0, getHeartbeatMax());
+    }
+    
+    protected void secure(final Connection conn, final byte[] response)
+    {
+        final SaslServer ss = conn.getSaslServer();
+        secure(ss, conn, response);
+    }
+
     protected int getHeartbeatMax()
     {
         return 0xFFFF;