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@1146079 13f79535-47bb-0310-9956-ffa450edef68

29 files changed, 470 insertions, 137 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
index 08eb05680c..8141533045 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java
index fa2fb9ead1..f21158cd0c 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/BrokerLink.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java
index 79de0678f0..09f35da41d 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
index 544bd62ed8..2dd9a63540 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
index 8939cfa334..0cfed77f2e 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
index f28873940b..9b357403a8 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java
index a697029d24..c1c836f9b4 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/subjects/ConnectionLogSubject.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java
index ce6bd3ee33..68f7689283 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/management/MBeanInvocationHandlerImpl.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java
index 3107185006..4e40305dbb 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
index 08f276ae72..d6201f7cf6 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
@@ -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/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
index c64ed4ad5a..9116bf2767 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java
index fcac78fafa..16d99de492 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
index 42a604e3a5..8c62441d59 100755
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
index 9b9de8333b..9140a13625 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
index 274cb6714a..4dbe94e4d7 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/security/AuthorizationHolder.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/AuthorizationHolder.java
new file mode 100755
index 0000000000..3d8c77a86f
--- /dev/null
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/security/SecurityManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/SecurityManager.java
index f18c327692..f582fed6a0 100755
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/SecurityManager.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java
index 70a9ea5356..e4bf8df340 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/access/ObjectProperties.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/GroupPrincipal.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/GroupPrincipal.java
new file mode 100644
index 0000000000..30a503c769
--- /dev/null
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java
index 6cc5e7b019..33aebffcfb 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
index 2113494cc5..9b3673e8b7 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
index 174dcbfa69..27e199291d 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
index 5c6206c53c..e9168f71fb 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
index be659c87ae..4b8b13fc7f 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
+++ b/qpid/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/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/UnitTestMessageLogger.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/UnitTestMessageLogger.java
index 3752dcb37e..e8defd0e58 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/UnitTestMessageLogger.java
+++ b/qpid/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/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java
index 2b724af2b1..3af665141c 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/InternalTestProtocolSession.java
+++ b/qpid/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/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
index 888a16053c..4c31092983 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
+++ b/qpid/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/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/GroupPrincipalTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/GroupPrincipalTest.java
new file mode 100644
index 0000000000..076b7c9248
--- /dev/null
+++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/security/PrincipalHolder.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/sasl/TestPrincipalUtils.java
index 7e93623cab..8b9b2df5a3 100755..100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/PrincipalHolder.java
+++ b/qpid/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;
+    }
+
 }