QPID-1807 : Add 0.5-fix broker and update SlowMessageStore to use MessageStores rather than TransactionLogs

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

47 files changed, 6505 insertions, 0 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLManager.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLManager.java
new file mode 100644
index 0000000000..6f7f66fad2
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLManager.java
@@ -0,0 +1,322 @@
+/*
+ *  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;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.log4j.Logger;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.configuration.SecurityConfiguration;
+import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.qpid.server.configuration.VirtualHostConfiguration;
+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.ACLPlugin.AuthzResult;
+import org.apache.qpid.server.security.access.plugins.SimpleXML;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+
+public class ACLManager
+{
+    private static final Logger _logger = Logger.getLogger(ACLManager.class);
+    private PluginManager _pluginManager;
+    private Map<String, ACLPluginFactory> _allSecurityPlugins = new HashMap<String, ACLPluginFactory>();
+    private Map<String, ACLPlugin> _globalPlugins = new HashMap<String, ACLPlugin>();
+    private Map<String, ACLPlugin> _hostPlugins = new HashMap<String, ACLPlugin>();
+
+    public ACLManager(SecurityConfiguration configuration, PluginManager manager) throws ConfigurationException
+    {
+        this(configuration, manager, null);
+    }
+
+    public ACLManager(SecurityConfiguration configuration, PluginManager manager, ACLPluginFactory securityPlugin) throws ConfigurationException
+    {
+        _pluginManager = manager;
+
+        if (manager == null) // No plugin manager, no plugins
+        {
+            return;
+        }
+
+        _allSecurityPlugins = _pluginManager.getSecurityPlugins();
+        if (securityPlugin != null)
+        {
+            _allSecurityPlugins.put(securityPlugin.getClass().getName(), securityPlugin);
+        }
+
+        _globalPlugins = configurePlugins(configuration);
+    }
+
+
+    public void configureHostPlugins(SecurityConfiguration hostConfig) throws ConfigurationException
+    {
+        _hostPlugins = configurePlugins(hostConfig);
+    }
+    
+    public Map<String, ACLPlugin> configurePlugins(SecurityConfiguration hostConfig) throws ConfigurationException
+    {
+        Configuration securityConfig = hostConfig.getConfiguration();
+        Map<String, ACLPlugin> plugins = new HashMap<String, ACLPlugin>();
+        Iterator keys = securityConfig.getKeys();
+        Collection<String> handledTags = new HashSet();
+        while (keys.hasNext())
+        {
+            // Splitting the string is necessary here because of the way that getKeys() returns only
+            // bottom level children
+            String tag = ((String) keys.next()).split("\\.", 2)[0];
+            if (!handledTags.contains(tag))
+            {
+                for (ACLPluginFactory plugin : _allSecurityPlugins.values())
+                {
+                    if (plugin.supportsTag(tag))
+                    {
+                        _logger.warn("Plugin handling security section "+tag+" is "+plugin.getClass().getSimpleName());
+                        handledTags.add(tag);
+                        plugins.put(plugin.getClass().getName(), plugin.newInstance(securityConfig));
+                    }
+                }
+            }
+            if (!handledTags.contains(tag))
+            {
+                _logger.warn("No plugin handled security section "+tag);
+            }
+        }
+        return plugins;
+    }    
+
+    public static Logger getLogger()
+    {
+        return _logger;
+    }
+
+    private abstract class AccessCheck
+    {
+        abstract AuthzResult allowed(ACLPlugin plugin);
+    }
+
+    private boolean checkAllPlugins(AccessCheck checker)
+    {
+        AuthzResult result = AuthzResult.ABSTAIN;
+        HashMap<String, ACLPlugin> remainingPlugins = new HashMap<String, ACLPlugin>();
+        remainingPlugins.putAll(_globalPlugins);
+        for (Entry<String, ACLPlugin> plugin : _hostPlugins.entrySet())
+        {
+            result = checker.allowed(plugin.getValue());
+            if (result == AuthzResult.DENIED)
+            {
+                // Something vetoed the access, we're done
+                return false; 
+            }
+            else if (result == AuthzResult.ALLOWED)
+            {
+                // Remove plugin from global check list since 
+                // host allow overrides global allow
+                remainingPlugins.remove(plugin.getKey());
+            }
+        }
+        
+        for (ACLPlugin plugin : remainingPlugins.values())
+        {   
+            result = checker.allowed(plugin);
+            if (result == AuthzResult.DENIED)
+            {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    public boolean authoriseBind(final AMQProtocolSession session, final Exchange exch, final AMQQueue queue,
+            final AMQShortString routingKey)
+    {
+        return checkAllPlugins(new AccessCheck()
+        {
+
+            @Override
+            AuthzResult allowed(ACLPlugin plugin)
+            {
+                return plugin.authoriseBind(session, exch, queue, routingKey);
+            }
+
+        });
+    }
+
+    public boolean authoriseConnect(final AMQProtocolSession session, final VirtualHost virtualHost)
+    {
+        return checkAllPlugins(new AccessCheck()
+        {
+
+            @Override
+            AuthzResult allowed(ACLPlugin plugin)
+            {
+                return plugin.authoriseConnect(session, virtualHost);
+            }
+
+        });
+    }
+
+    public boolean authoriseConsume(final AMQProtocolSession session, final boolean noAck, final AMQQueue queue)
+    {
+        return checkAllPlugins(new AccessCheck()
+        {
+
+            @Override
+            AuthzResult allowed(ACLPlugin plugin)
+            {
+                return plugin.authoriseConsume(session, noAck, queue);
+            }
+
+        });
+    }
+
+    public boolean authoriseConsume(final AMQProtocolSession session, final boolean exclusive, final boolean noAck,
+            final boolean noLocal, final boolean nowait, final AMQQueue queue)
+    {
+        return checkAllPlugins(new AccessCheck()
+        {
+
+            @Override
+            AuthzResult allowed(ACLPlugin plugin)
+            {
+                return plugin.authoriseConsume(session, exclusive, noAck, noLocal, nowait, queue);
+            }
+
+        });
+    }
+
+    public boolean authoriseCreateExchange(final AMQProtocolSession session, final boolean autoDelete,
+            final boolean durable, final AMQShortString exchangeName, final boolean internal, final boolean nowait,
+            final boolean passive, final AMQShortString exchangeType)
+    {
+        return checkAllPlugins(new AccessCheck()
+        {
+
+            @Override
+            AuthzResult allowed(ACLPlugin plugin)
+            {
+                return plugin.authoriseCreateExchange(session, autoDelete, durable, exchangeName, internal, nowait,
+                        passive, exchangeType);
+            }
+
+        });
+    }
+
+    public boolean authoriseCreateQueue(final AMQProtocolSession session, final boolean autoDelete,
+            final boolean durable, final boolean exclusive, final boolean nowait, final boolean passive,
+            final AMQShortString queue)
+    {
+        return checkAllPlugins(new AccessCheck()
+        {
+
+            @Override
+            AuthzResult allowed(ACLPlugin plugin)
+            {
+                return plugin.authoriseCreateQueue(session, autoDelete, durable, exclusive, nowait, passive, queue);
+            }
+
+        });
+    }
+
+    public boolean authoriseDelete(final AMQProtocolSession session, final AMQQueue queue)
+    {
+        return checkAllPlugins(new AccessCheck()
+        {
+
+            @Override
+            AuthzResult allowed(ACLPlugin plugin)
+            {
+                return plugin.authoriseDelete(session, queue);
+            }
+
+        });
+    }
+
+    public boolean authoriseDelete(final AMQProtocolSession session, final Exchange exchange)
+    {
+        return checkAllPlugins(new AccessCheck()
+        {
+
+            @Override
+            AuthzResult allowed(ACLPlugin plugin)
+            {
+                return plugin.authoriseDelete(session, exchange);
+            }
+
+        });
+    }
+    
+    public boolean authorisePublish(final AMQProtocolSession session, final boolean immediate, final boolean mandatory,
+            final AMQShortString routingKey, final Exchange e)
+    {
+        return checkAllPlugins(new AccessCheck()
+        {
+
+            @Override
+            AuthzResult allowed(ACLPlugin plugin)
+            {
+                return plugin.authorisePublish(session, immediate, mandatory, routingKey, e);
+            }
+
+        });
+    }
+
+    public boolean authorisePurge(final AMQProtocolSession session, final AMQQueue queue)
+    {
+        return checkAllPlugins(new AccessCheck()
+        {
+
+            @Override
+            AuthzResult allowed(ACLPlugin plugin)
+            {
+                return plugin.authorisePurge(session, queue);
+            }
+
+        });
+    }
+
+    public boolean authoriseUnbind(final AMQProtocolSession session, final Exchange exch,
+            final AMQShortString routingKey, final AMQQueue queue)
+    {
+        return checkAllPlugins(new AccessCheck()
+        {
+
+            @Override
+            AuthzResult allowed(ACLPlugin plugin)
+            {
+                return plugin.authoriseUnbind(session, exch, routingKey, queue);
+            }
+
+        });
+    }
+
+    public void addHostPlugin(ACLPlugin aclPlugin)
+    {
+        _hostPlugins.put(aclPlugin.getClass().getName(), aclPlugin);
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLPlugin.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLPlugin.java
new file mode 100644
index 0000000000..032184ec39
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLPlugin.java
@@ -0,0 +1,70 @@
+/*
+ *  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;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+
+public interface ACLPlugin
+{
+    public enum AuthzResult
+    {
+        ALLOWED,
+        DENIED,
+        ABSTAIN        
+    }
+
+    void setConfiguration(Configuration config) throws ConfigurationException;
+
+    // These return true if the plugin thinks the action should be allowed, and false if not. 
+    
+    AuthzResult authoriseBind(AMQProtocolSession session, Exchange exch, AMQQueue queue, AMQShortString routingKey);
+
+    AuthzResult authoriseCreateExchange(AMQProtocolSession session, boolean autoDelete, boolean durable,
+            AMQShortString exchangeName, boolean internal, boolean nowait, boolean passive, AMQShortString exchangeType);
+
+    AuthzResult authoriseCreateQueue(AMQProtocolSession session, boolean autoDelete, boolean durable, boolean exclusive,
+            boolean nowait, boolean passive, AMQShortString queue);
+
+    AuthzResult authoriseConnect(AMQProtocolSession session, VirtualHost virtualHost);
+
+    AuthzResult authoriseConsume(AMQProtocolSession session, boolean noAck, AMQQueue queue);
+
+    AuthzResult authoriseConsume(AMQProtocolSession session, boolean exclusive, boolean noAck, boolean noLocal,
+            boolean nowait, AMQQueue queue);
+
+    AuthzResult authoriseDelete(AMQProtocolSession session, AMQQueue queue);
+
+    AuthzResult authoriseDelete(AMQProtocolSession session, Exchange exchange);
+
+    AuthzResult authorisePublish(AMQProtocolSession session, boolean immediate, boolean mandatory,
+            AMQShortString routingKey, Exchange e);
+
+    AuthzResult authorisePurge(AMQProtocolSession session, AMQQueue queue);
+
+    AuthzResult authoriseUnbind(AMQProtocolSession session, Exchange exch, AMQShortString routingKey, AMQQueue queue);
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLPluginFactory.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLPluginFactory.java
new file mode 100644
index 0000000000..256f093477
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/ACLPluginFactory.java
@@ -0,0 +1,33 @@
+/*
+ *
+ * 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;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+
+public interface ACLPluginFactory
+{
+
+    public boolean supportsTag(String name);
+    
+    public ACLPlugin newInstance(Configuration config) throws ConfigurationException;
+    
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessResult.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessResult.java
new file mode 100644
index 0000000000..d722da4ae0
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessResult.java
@@ -0,0 +1,65 @@
+/*
+ *  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;
+
+public class AccessResult
+{
+    public enum AccessStatus
+    {
+        GRANTED, REFUSED
+    }
+
+    private String _authorizer;
+    private AccessStatus _status;
+
+    public AccessResult(ACLPlugin authorizer, AccessStatus status)
+    {
+        _status = status;
+        _authorizer = authorizer.getClass().getSimpleName();
+    }
+
+    public void setAuthorizer(ACLPlugin authorizer)
+    {
+        _authorizer += authorizer.getClass().getSimpleName();
+    }
+
+    public String getAuthorizer()
+    {
+        return _authorizer;
+    }
+
+    public void setStatus(AccessStatus status)
+    {
+        _status = status;
+    }
+
+    public AccessStatus getStatus()
+    {
+        return _status;
+    }
+
+    public void addAuthorizer(ACLPlugin accessManager)
+    {
+        _authorizer = accessManager.getClass().getSimpleName() + "->" + _authorizer;
+    }
+
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessRights.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessRights.java
new file mode 100644
index 0000000000..1b79a5a0e0
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/AccessRights.java
@@ -0,0 +1,63 @@
+/*
+ *  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;
+
+public class AccessRights
+{
+    public enum Rights
+    {
+        ANY,
+        READ,
+        WRITE,
+        READWRITE
+    }
+
+    Rights _right;
+
+    public AccessRights(Rights right)
+    {
+        _right = right;
+    }
+
+    public boolean allows(Rights rights)
+    {
+        switch (_right)
+        {
+            case ANY:
+                return (rights.equals(Rights.WRITE)
+                        || rights.equals(Rights.READ)
+                        || rights.equals(Rights.READWRITE)
+                        || rights.equals(Rights.ANY));
+            case READ:
+                return rights.equals(Rights.READ) || rights.equals(Rights.ANY);
+            case WRITE:
+                return rights.equals(Rights.WRITE) || rights.equals(Rights.ANY);
+            case READWRITE:
+                return true;
+        }
+        return false;
+    }
+
+    public Rights getRights()
+    {
+        return _right;
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/Accessable.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/Accessable.java
new file mode 100644
index 0000000000..f51cf24caa
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/Accessable.java
@@ -0,0 +1,27 @@
+/*
+ *  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;
+
+public interface Accessable
+{
+    void setAccessableName(String name);
+    String getAccessableName();    
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/AuthorizationManager.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/AuthorizationManager.java
new file mode 100644
index 0000000000..9527120f30
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/AuthorizationManager.java
@@ -0,0 +1,6 @@
+package org.apache.qpid.server.security.access;
+
+public class AuthorizationManager
+{
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/Permission.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/Permission.java
new file mode 100644
index 0000000000..b65b0cdc6c
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/Permission.java
@@ -0,0 +1,38 @@
+/*
+ *  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;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.queue.AMQQueue;
+
+public enum Permission
+{
+    CONSUME,
+    PUBLISH,
+    CREATEQUEUE,
+    CREATEEXCHANGE,
+    ACCESS,
+    BIND,
+    UNBIND,
+    DELETE,
+    PURGE
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/PrincipalPermissions.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/PrincipalPermissions.java
new file mode 100755
index 0000000000..f852514444
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/PrincipalPermissions.java
@@ -0,0 +1,612 @@
+/*
+ *
+ * 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;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.QueueBindBody;
+import org.apache.qpid.framing.QueueDeclareBody;
+import org.apache.qpid.framing.ExchangeDeclareBody;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.security.access.ACLPlugin.AuthzResult;
+import org.apache.qpid.server.exchange.Exchange;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class PrincipalPermissions
+{
+
+    private static final Object CONSUME_QUEUES_KEY = new Object();
+    private static final Object CONSUME_TEMPORARY_KEY = new Object();
+    private static final Object CONSUME_OWN_QUEUES_ONLY_KEY = new Object();
+
+    private static final Object CREATE_QUEUES_KEY = new Object();
+    private static final Object CREATE_EXCHANGES_KEY = new Object();
+
+    private static final Object CREATE_QUEUE_TEMPORARY_KEY = new Object();
+    private static final Object CREATE_QUEUE_QUEUES_KEY = new Object();
+    private static final Object CREATE_QUEUE_EXCHANGES_KEY = new Object();
+
+    private static final Object CREATE_QUEUE_EXCHANGES_TEMPORARY_KEY = new Object();
+    private static final Object CREATE_QUEUE_EXCHANGES_ROUTINGKEYS_KEY = new Object();
+
+    private static final int PUBLISH_EXCHANGES_KEY = 0;
+
+    private Map _permissions;
+
+    private String _user;
+
+
+    public PrincipalPermissions(String user)
+    {
+        _user = user;
+        _permissions = new ConcurrentHashMap();
+    }
+
+    /**
+     * 
+     * @param permission the type of permission to check
+     * 
+     * @param parameters vararg depending on what permission was passed in
+     *  ACCESS: none
+     *  BIND: none
+     *  CONSUME: AMQShortString queueName, Boolean temporary, Boolean ownQueueOnly
+     *  CREATEQUEUE:  Boolean temporary, AMQShortString queueName, AMQShortString exchangeName, AMQShortString routingKey
+     *  CREATEEXCHANGE: AMQShortString exchangeName, AMQShortString Class
+     *  DELETE: none
+     *  PUBLISH: Exchange exchange, AMQShortString routingKey
+     *  PURGE: none
+     *  UNBIND: none
+     */
+    public void grant(Permission permission, Object... parameters)
+    {
+        switch (permission)
+        {
+            case ACCESS:
+                break; // This is a no-op as the existence of this PrincipalPermission object is scoped per VHost for ACCESS
+            case BIND:
+                break; // All the details are currently included in the create setup.
+            case CONSUME: // Parameters : AMQShortString queueName, Boolean Temporary, Boolean ownQueueOnly
+                Map consumeRights = (Map) _permissions.get(permission);
+
+                if (consumeRights == null)
+                {
+                    consumeRights = new ConcurrentHashMap();
+                    _permissions.put(permission, consumeRights);
+                }
+
+                //if we have parametsre
+                if (parameters.length > 0)
+                {
+                    AMQShortString queueName = (AMQShortString) parameters[0];
+                    Boolean temporary = (Boolean) parameters[1];
+                    Boolean ownQueueOnly = (Boolean) parameters[2];
+
+                    if (temporary)
+                    {
+                        consumeRights.put(CONSUME_TEMPORARY_KEY, true);
+                    }
+                    else
+                    {
+                        consumeRights.put(CONSUME_TEMPORARY_KEY, false);
+                    }
+
+                    if (ownQueueOnly)
+                    {
+                        consumeRights.put(CONSUME_OWN_QUEUES_ONLY_KEY, true);
+                    }
+                    else
+                    {
+                        consumeRights.put(CONSUME_OWN_QUEUES_ONLY_KEY, false);
+                    }
+
+
+                    LinkedList queues = (LinkedList) consumeRights.get(CONSUME_QUEUES_KEY);
+                    if (queues == null)
+                    {
+                        queues = new LinkedList();
+                        consumeRights.put(CONSUME_QUEUES_KEY, queues);
+                    }
+
+                    if (queueName != null)
+                    {
+                        queues.add(queueName);
+                    }
+                }
+
+
+                break;
+            case CREATEQUEUE:  // Parameters : Boolean temporary, AMQShortString queueName
+                // , AMQShortString exchangeName , AMQShortString routingKey
+
+                Map createRights = (Map) _permissions.get(permission);
+
+                if (createRights == null)
+                {
+                    createRights = new ConcurrentHashMap();
+                    _permissions.put(permission, createRights);
+
+                }
+
+                //The existence of the empty map mean permission to all.
+                if (parameters.length == 0)
+                {
+                    return;
+                }
+
+                Boolean temporary = (Boolean) parameters[0];
+
+                AMQShortString queueName = parameters.length > 1 ? (AMQShortString) parameters[1] : null;
+                AMQShortString exchangeName = parameters.length > 2 ? (AMQShortString) parameters[2] : null;
+                //Set the routingkey to the specified value or the queueName if present
+                AMQShortString routingKey = (parameters.length > 3 && null != parameters[3]) ? (AMQShortString) parameters[3] : queueName;
+
+                // Get the queues map
+                Map create_queues = (Map) createRights.get(CREATE_QUEUES_KEY);
+
+                if (create_queues == null)
+                {
+                    create_queues = new ConcurrentHashMap();
+                    createRights.put(CREATE_QUEUES_KEY, create_queues);
+                }
+
+                //Allow all temp queues to be created
+                create_queues.put(CREATE_QUEUE_TEMPORARY_KEY, temporary);
+
+                //Create empty list of queues
+                Map create_queues_queues = (Map) create_queues.get(CREATE_QUEUE_QUEUES_KEY);
+
+                if (create_queues_queues == null)
+                {
+                    create_queues_queues = new ConcurrentHashMap();
+                    create_queues.put(CREATE_QUEUE_QUEUES_KEY, create_queues_queues);
+                }
+
+                // We are granting CREATE rights to all temporary queues only
+                if (parameters.length == 1)
+                {
+                    return;
+                }
+
+                // if we have a queueName then we need to store any associated exchange / rk bindings
+                if (queueName != null)
+                {
+                    Map queue = (Map) create_queues_queues.get(queueName);
+                    if (queue == null)
+                    {
+                        queue = new ConcurrentHashMap();
+                        create_queues_queues.put(queueName, queue);
+                    }
+
+                    if (exchangeName != null)
+                    {
+                        queue.put(exchangeName, routingKey);
+                    }
+
+                    //If no exchange is specified then the presence of the queueName in the map says any exchange is ok
+                }
+
+                // Store the exchange that we are being granted rights to. This will be used as part of binding
+
+                //Lookup the list of exchanges
+                Map create_queues_exchanges = (Map) create_queues.get(CREATE_QUEUE_EXCHANGES_KEY);
+
+                if (create_queues_exchanges == null)
+                {
+                    create_queues_exchanges = new ConcurrentHashMap();
+                    create_queues.put(CREATE_QUEUE_EXCHANGES_KEY, create_queues_exchanges);
+                }
+
+                //if we have an exchange
+                if (exchangeName != null)
+                {
+                    //Retrieve the list of permitted exchanges.
+                    Map exchanges = (Map) create_queues_exchanges.get(exchangeName);
+
+                    if (exchanges == null)
+                    {
+                        exchanges = new ConcurrentHashMap();
+                        create_queues_exchanges.put(exchangeName, exchanges);
+                    }
+
+                    //Store the temporary setting CREATE_QUEUE_EXCHANGES_ROUTINGKEYS_KEY
+                    exchanges.put(CREATE_QUEUE_EXCHANGES_TEMPORARY_KEY, temporary);
+
+                    //Store the binding details of queue/rk for this exchange.
+                    if (queueName != null)
+                    {
+                        //Retrieve the list of permitted routingKeys.
+                        Map rKeys = (Map) exchanges.get(exchangeName);
+
+                        if (rKeys == null)
+                        {
+                            rKeys = new ConcurrentHashMap();
+                            exchanges.put(CREATE_QUEUE_EXCHANGES_ROUTINGKEYS_KEY, rKeys);
+                        }
+
+                        rKeys.put(queueName, routingKey);
+                    }
+                }
+                break;
+            case CREATEEXCHANGE:
+                // Parameters AMQShortString exchangeName , AMQShortString Class
+                Map rights = (Map) _permissions.get(permission);
+                if (rights == null)
+                {
+                    rights = new ConcurrentHashMap();
+                    _permissions.put(permission, rights);
+                }
+
+                Map create_exchanges = (Map) rights.get(CREATE_EXCHANGES_KEY);
+                if (create_exchanges == null)
+                {
+                    create_exchanges = new ConcurrentHashMap();
+                    rights.put(CREATE_EXCHANGES_KEY, create_exchanges);
+                }
+
+                //Should perhaps error if parameters[0] is null;
+                AMQShortString name = parameters.length > 0 ? (AMQShortString) parameters[0] : null;
+                AMQShortString className = parameters.length > 1 ? (AMQShortString) parameters[1] : new AMQShortString("direct");
+
+                //Store the exchangeName / class mapping if the mapping is null
+                rights.put(name, className);
+                break;
+            case DELETE:
+                break;
+
+            case PUBLISH: // Parameters : Exchange exchange, AMQShortString routingKey
+                Map publishRights = (Map) _permissions.get(permission);
+
+                if (publishRights == null)
+                {
+                    publishRights = new ConcurrentHashMap();
+                    _permissions.put(permission, publishRights);
+                }
+
+                if (parameters == null || parameters.length == 0)
+                {
+                    //If we have no parameters then allow publish to all destinations
+                    // this is signified by having a null value for publish_exchanges
+                }
+                else
+                {
+                    Map publish_exchanges = (Map) publishRights.get(PUBLISH_EXCHANGES_KEY);
+
+                    if (publish_exchanges == null)
+                    {
+                        publish_exchanges = new ConcurrentHashMap();
+                        publishRights.put(PUBLISH_EXCHANGES_KEY, publish_exchanges);
+                    }
+
+
+                    HashSet routingKeys = (HashSet) publish_exchanges.get(parameters[0]);
+
+                    // Check to see if we have a routing key
+                    if (parameters.length == 2)
+                    {
+                        if (routingKeys == null)
+                        {
+                            routingKeys = new HashSet<AMQShortString>();
+                        }
+                        //Add routing key to permitted publish destinations
+                        routingKeys.add(parameters[1]);
+                    }
+
+                    // Add the updated routingkey list or null if all values allowed
+                    publish_exchanges.put(parameters[0], routingKeys);
+                }
+                break;
+            case PURGE:
+                break;
+            case UNBIND:
+                break;
+        }
+
+    }
+
+    /**
+     * 
+     * @param permission the type of permission to check
+     * 
+     * @param parameters vararg depending on what permission was passed in
+     *  ACCESS: none
+     *  BIND: QueueBindBody bindmethod, Exchange exchange, AMQQueue queue, AMQShortString routingKey
+     *  CONSUME: AMQQueue queue
+     *  CREATEQUEUE:  Boolean autodelete, AMQShortString name
+     *  CREATEEXCHANGE: AMQShortString exchangeName
+     *  DELETE: none
+     *  PUBLISH: Exchange exchange, AMQShortString routingKey
+     *  PURGE: none
+     *  UNBIND: none
+     */
+    public AuthzResult authorise(Permission permission, Object... parameters)
+    {
+
+        switch (permission)
+        {
+            case ACCESS:
+                return AuthzResult.ALLOWED; // This is here for completeness but the SimpleXML ACLManager never calls it.
+                // The existence of this user specific PP can be validated in the map SimpleXML maintains.
+            case BIND: // Parameters : QueueBindMethod , Exchange , AMQQueue, AMQShortString routingKey
+
+                Exchange exchange = (Exchange) parameters[1];
+
+                AMQQueue bind_queueName = (AMQQueue) parameters[2];
+                AMQShortString routingKey = (AMQShortString) parameters[3];
+
+                //Get all Create Rights for this user
+                Map bindCreateRights = (Map) _permissions.get(Permission.CREATEQUEUE);
+
+                //Look up the Queue Creation Rights
+                Map bind_create_queues = (Map) bindCreateRights.get(CREATE_QUEUES_KEY);
+
+                //Lookup the list of queues
+                Map bind_create_queues_queues = (Map) bindCreateRights.get(CREATE_QUEUE_QUEUES_KEY);
+
+                // Check and see if we have a queue white list to check
+                if (bind_create_queues_queues != null)
+                {
+                    //There a white list for queues
+                    Map exchangeDetails = (Map) bind_create_queues_queues.get(bind_queueName);
+
+                    if (exchangeDetails == null) //Then all queue can be bound to all exchanges.
+                    {
+                        return AuthzResult.ALLOWED;
+                    }
+
+                    // Check to see if we have a white list of routingkeys to check
+                    Map rkeys = (Map) exchangeDetails.get(exchange.getName());
+
+                    // if keys is null then any rkey is allowed on this exchange
+                    if (rkeys == null)
+                    {
+                        // There is no routingkey white list
+                        return AuthzResult.ALLOWED;
+                    }
+                    else
+                    {
+                        // We have routingKeys so a match must be found to allowed binding
+                        Iterator keys = rkeys.keySet().iterator();
+
+                        boolean matched = false;
+                        while (keys.hasNext() && !matched)
+                        {
+                            AMQShortString rkey = (AMQShortString) keys.next();
+                            if (rkey.endsWith("*"))
+                            {
+                                matched = routingKey.startsWith(rkey.subSequence(0, rkey.length() - 1).toString());
+                            }
+                            else
+                            {
+                                matched = routingKey.equals(rkey);
+                            }
+                        }
+
+
+                        return (matched) ? AuthzResult.ALLOWED : AuthzResult.DENIED;
+                    }
+
+
+                }
+                else
+                {
+                    //There a is no white list for queues
+
+                    // So can allow all queues to be bound
+                    //  but we should first check and see if we have a temp queue and validate that we are allowed
+                    //  to bind temp queues.
+
+                    //Check to see if we have a temporary queue
+                    if (bind_queueName.isAutoDelete())
+                    {
+                        // Check and see if we have an exchange white list.
+                        Map bind_exchanges = (Map) bind_create_queues.get(CREATE_QUEUE_EXCHANGES_KEY);
+
+                        // If the exchange exists then we must check to see if temporary queues are allowed here
+                        if (bind_exchanges != null)
+                        {
+                            // Check to see if the requested exchange is allowed.
+                            Map exchangeDetails = (Map) bind_exchanges.get(exchange.getName());
+
+                            return ((Boolean) exchangeDetails.get(CREATE_QUEUE_EXCHANGES_TEMPORARY_KEY)) ? AuthzResult.ALLOWED : AuthzResult.DENIED;
+                        }
+
+                        //no white list so all allowed, drop through to return true below.
+                    }
+
+                    // not a temporary queue and no white list so all allowed.
+                    return AuthzResult.ALLOWED;
+                }
+
+            case CREATEQUEUE:// Parameters : boolean autodelete, AMQShortString name
+
+                Map createRights = (Map) _permissions.get(permission);
+
+                // If there are no create rights then deny request
+                if (createRights == null)
+                {
+                    return AuthzResult.DENIED;
+                }
+
+                //Look up the Queue Creation Rights
+                Map create_queues = (Map) createRights.get(CREATE_QUEUES_KEY);
+
+                //Lookup the list of queues allowed to be created
+                Map create_queues_queues = (Map) create_queues.get(CREATE_QUEUE_QUEUES_KEY);
+
+
+                AMQShortString queueName = (AMQShortString) parameters[1];
+                Boolean autoDelete = (Boolean) parameters[0];
+
+                if (autoDelete)// we have a temporary queue
+                {
+                    return ((Boolean) create_queues.get(CREATE_QUEUE_TEMPORARY_KEY)) ? AuthzResult.ALLOWED : AuthzResult.DENIED;
+                }
+                else
+                {
+                    // If there is a white list then check
+                    if (create_queues_queues == null || create_queues_queues.containsKey(queueName))
+                    {
+                        return AuthzResult.ALLOWED; 
+                    }
+                    else
+                    {
+                        return AuthzResult.DENIED;
+                    }
+                        
+                }
+            case CREATEEXCHANGE:
+                Map rights = (Map) _permissions.get(permission);
+
+                AMQShortString exchangeName = (AMQShortString) parameters[0];
+
+                // If the exchange list is doesn't exist then all is allowed else
+                // check the valid exchanges
+                if (rights == null || rights.containsKey(exchangeName))
+                {
+                    return AuthzResult.ALLOWED; 
+                }
+                else
+                {
+                    return AuthzResult.DENIED;
+                }
+            case CONSUME: // Parameters :  AMQQueue
+
+                if (parameters.length == 1 && parameters[0] instanceof AMQQueue)
+                {
+                    AMQQueue queue = ((AMQQueue) parameters[0]);
+                    Map queuePermissions = (Map) _permissions.get(permission);
+
+                    List queues = (List) queuePermissions.get(CONSUME_QUEUES_KEY);
+
+                    Boolean temporayQueues = (Boolean) queuePermissions.get(CONSUME_TEMPORARY_KEY);
+                    Boolean ownQueuesOnly = (Boolean) queuePermissions.get(CONSUME_OWN_QUEUES_ONLY_KEY);
+
+                    // If user is allowed to publish to temporary queues and this is a temp queue then allow it.
+                    if (temporayQueues)
+                    {
+                        if (queue.isAutoDelete())
+                        // This will allow consumption from any temporary queue including ones not owned by this user.
+                        // Of course the exclusivity will not be broken.
+                        {
+                            // if not limited to ownQueuesOnly then ok else check queue Owner.
+                            return (!ownQueuesOnly || queue.getOwner().equals(_user)) ? AuthzResult.ALLOWED : AuthzResult.DENIED;
+                        }
+                        else
+                        {
+                            return AuthzResult.DENIED;
+                        }
+                    }
+
+                    // if queues are white listed then ensure it is ok
+                    if (queues != null)
+                    {
+                        // if no queues are listed then ALL are ok othereise it must be specified.
+                        if (ownQueuesOnly)
+                        {
+                            if (queue.getOwner().equals(_user))
+                            {
+                                return (queues.size() == 0 || queues.contains(queue.getName())) ? AuthzResult.ALLOWED : AuthzResult.DENIED;
+                            }
+                            else
+                            {
+                                return AuthzResult.DENIED;
+                            }
+                        }
+
+                        // If we are
+                        return (queues.size() == 0 || queues.contains(queue.getName())) ? AuthzResult.ALLOWED : AuthzResult.DENIED;
+                    }
+                }
+
+                // Can't authenticate without the right parameters
+                return AuthzResult.DENIED;
+            case DELETE:
+                break;
+
+            case PUBLISH: // Parameters : Exchange exchange, AMQShortString routingKey
+                Map publishRights = (Map) _permissions.get(permission);
+
+                if (publishRights == null)
+                {
+                    return AuthzResult.DENIED;
+                }
+
+                Map exchanges = (Map) publishRights.get(PUBLISH_EXCHANGES_KEY);
+
+                // Having no exchanges listed gives full publish rights to all exchanges
+                if (exchanges == null)
+                {
+                    return AuthzResult.ALLOWED;
+                }
+                // Otherwise exchange must be listed in the white list
+
+                // If the map doesn't have the exchange then it isn't allowed
+                if (!exchanges.containsKey(((Exchange) parameters[0]).getName()))
+                {
+                    return AuthzResult.DENIED;
+                }
+                else
+                {
+
+                    // Get valid routing keys
+                    HashSet routingKeys = (HashSet) exchanges.get(((Exchange)parameters[0]).getName());
+
+                    // Having no routingKeys in the map then all are allowed.
+                    if (routingKeys == null)
+                    {
+                        return AuthzResult.ALLOWED;
+                    }
+                    else
+                    {
+                        // We have routingKeys so a match must be found to allowed binding
+                        Iterator keys = routingKeys.iterator();
+
+
+                        AMQShortString publishRKey = (AMQShortString)parameters[1];
+
+                        boolean matched = false;
+                        while (keys.hasNext() && !matched)
+                        {
+                            AMQShortString rkey = (AMQShortString) keys.next();
+
+                            if (rkey.endsWith("*"))
+                            {
+                                matched = publishRKey.startsWith(rkey.subSequence(0, rkey.length() - 1));
+                            }
+                            else
+                            {
+                                matched = publishRKey.equals(rkey);
+                            }
+                        }
+                        return (matched) ? AuthzResult.ALLOWED : AuthzResult.DENIED;
+                    }
+                }
+            case PURGE:
+                break;
+            case UNBIND:
+                break;
+
+        }
+
+        return AuthzResult.DENIED;
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/VirtualHostAccess.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/VirtualHostAccess.java
new file mode 100644
index 0000000000..13151a66b8
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/VirtualHostAccess.java
@@ -0,0 +1,68 @@
+/*
+ *  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;
+
+public class VirtualHostAccess
+{
+    private String _vhost;
+    private AccessRights _rights;
+
+    public VirtualHostAccess(String vhostaccess)
+    {
+        //format <vhost>(<rights>)
+        int hostend = vhostaccess.indexOf('(');
+
+        if (hostend == -1)
+        {
+            throw new IllegalArgumentException("VirtualHostAccess format string contains no access _rights");
+        }
+
+        _vhost = vhostaccess.substring(0, hostend);
+
+        String rights = vhostaccess.substring(hostend);
+
+        if (rights.indexOf('r') != -1)
+        {
+            if (rights.indexOf('w') != -1)
+            {
+                _rights = new AccessRights(AccessRights.Rights.READWRITE);
+            }
+            else
+            {
+                _rights = new AccessRights(AccessRights.Rights.READ);
+            }
+        }
+        else if (rights.indexOf('w') != -1)
+        {
+            _rights = new AccessRights(AccessRights.Rights.WRITE);
+        }
+    }
+
+    public AccessRights getAccessRights()
+    {
+        return _rights;
+    }
+
+    public String getVirtualHost()
+    {
+        return _vhost;
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/management/AMQUserManagementMBean.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/management/AMQUserManagementMBean.java
new file mode 100644
index 0000000000..121f571abe
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/management/AMQUserManagementMBean.java
@@ -0,0 +1,501 @@
+/*
+ *  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.management;
+
+import org.apache.qpid.server.management.MBeanDescription;
+import org.apache.qpid.server.management.AMQManagedObject;
+import org.apache.qpid.server.management.MBeanOperation;
+import org.apache.qpid.server.management.MBeanInvocationHandlerImpl;
+import org.apache.qpid.server.security.auth.database.PrincipalDatabase;
+import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
+import org.apache.qpid.server.security.access.management.UserManagement;
+import org.apache.log4j.Logger;
+import org.apache.commons.configuration.ConfigurationException;
+
+import javax.management.JMException;
+import javax.management.remote.JMXPrincipal;
+import javax.management.openmbean.TabularData;
+import javax.management.openmbean.TabularDataSupport;
+import javax.management.openmbean.TabularType;
+import javax.management.openmbean.SimpleType;
+import javax.management.openmbean.CompositeType;
+import javax.management.openmbean.OpenType;
+import javax.management.openmbean.OpenDataException;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.CompositeDataSupport;
+import javax.security.auth.login.AccountNotFoundException;
+import javax.security.auth.Subject;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.FileOutputStream;
+import java.util.Properties;
+import java.util.List;
+import java.util.Enumeration;
+import java.util.Set;
+import java.util.concurrent.locks.ReentrantLock;
+import java.security.Principal;
+import java.security.AccessControlContext;
+import java.security.AccessController;
+
+/** MBean class for AMQUserManagementMBean. It implements all the management features exposed for managing users. */
+@MBeanDescription("User Management Interface")
+public class AMQUserManagementMBean extends AMQManagedObject implements UserManagement
+{
+
+    private static final Logger _logger = Logger.getLogger(AMQUserManagementMBean.class);
+
+    private PrincipalDatabase _principalDatabase;
+    private Properties _accessRights;
+    private File _accessFile;
+
+    private ReentrantLock _accessRightsUpdate = new ReentrantLock();
+
+    // Setup for the TabularType
+    static TabularType _userlistDataType; // Datatype for representing User Lists
+
+    static CompositeType _userDataType; // Composite type for representing User
+    static String[] _userItemNames = {"Username", "read", "write", "admin"};
+
+    static
+    {
+        String[] userItemDesc = {"Broker Login username", "Management Console Read Permission",
+                                 "Management Console Write Permission", "Management Console Admin Permission"};
+
+        OpenType[] userItemTypes = new OpenType[4]; // User item types.
+        userItemTypes[0] = SimpleType.STRING;  // For Username
+        userItemTypes[1] = SimpleType.BOOLEAN; // For Rights - Read
+        userItemTypes[2] = SimpleType.BOOLEAN; // For Rights - Write
+        userItemTypes[3] = SimpleType.BOOLEAN; // For Rights - Admin
+        String[] userDataIndex = {_userItemNames[0]};
+
+        try
+        {
+            _userDataType =
+                    new CompositeType("User", "User Data", _userItemNames, userItemDesc, userItemTypes);
+
+            _userlistDataType = new TabularType("Users", "List of users", _userDataType, userDataIndex);
+        }
+        catch (OpenDataException e)
+        {
+            _logger.error("Tabular data setup for viewing users incorrect.");
+            _userlistDataType = null;
+        }
+    }
+
+
+    public AMQUserManagementMBean() throws JMException
+    {
+        super(UserManagement.class, UserManagement.TYPE, UserManagement.VERSION);
+    }
+
+    public String getObjectInstanceName()
+    {
+        return UserManagement.TYPE;
+    }
+
+    public boolean setPassword(String username, char[] password)
+    {
+        try
+        {
+            //delegate password changes to the Principal Database
+            return _principalDatabase.updatePassword(new UsernamePrincipal(username), password);
+        }
+        catch (AccountNotFoundException e)
+        {
+            _logger.warn("Attempt to set password of non-existant user'" + username + "'");
+            return false;
+        }
+    }
+
+    public boolean setRights(String username, boolean read, boolean write, boolean admin)
+    {
+
+        Object oldRights = null;
+        if ((oldRights =_accessRights.get(username)) == null)
+        {
+            // If the user doesn't exist in the access rights file check that they at least have an account.
+            if (_principalDatabase.getUser(username) == null)
+            {
+                return false;
+            }
+        }
+
+        try
+        {
+            _accessRightsUpdate.lock();
+
+            // Update the access rights
+            if (admin)
+            {
+                _accessRights.put(username, MBeanInvocationHandlerImpl.ADMIN);
+            }
+            else
+            {
+                if (read | write)
+                {
+                    if (read)
+                    {
+                        _accessRights.put(username, MBeanInvocationHandlerImpl.READONLY);
+                    }
+                    if (write)
+                    {
+                        _accessRights.put(username, MBeanInvocationHandlerImpl.READWRITE);
+                    }
+                }
+                else
+                {
+                    _accessRights.remove(username);
+                }
+            }
+            
+            //save the rights file
+            try
+            {
+                saveAccessFile();
+            }
+            catch (IOException e)
+            {
+                _logger.warn("Problem occured saving '" + _accessFile + "', the access right changes will not be preserved: " + e);
+
+                //the rights file was not successfully saved, restore user rights to previous value
+                _logger.warn("Reverting attempted rights update for user'" + username + "'");
+                if (oldRights != null)
+                {
+                    _accessRights.put(username, oldRights);
+                }
+                else
+                {
+                    _accessRights.remove(username);
+                }
+
+                return false;
+            }
+        }
+        finally
+        {
+            if (_accessRightsUpdate.isHeldByCurrentThread())
+            {
+                _accessRightsUpdate.unlock();
+            }
+        }
+
+        return true;
+    }
+
+    public boolean createUser(String username, char[] password, boolean read, boolean write, boolean admin)
+    {
+        if (_principalDatabase.createPrincipal(new UsernamePrincipal(username), password))
+        {
+            if (!setRights(username, read, write, admin))
+            {
+                //unable to set rights for user, remove account
+                try
+                {
+                    _principalDatabase.deletePrincipal(new UsernamePrincipal(username));
+                }
+                catch (AccountNotFoundException e)
+                {
+                    //ignore
+                }
+                return false;
+            }
+            else
+            {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    public boolean deleteUser(String username)
+    {
+        try
+        {
+            if (_principalDatabase.deletePrincipal(new UsernamePrincipal(username)))
+            {
+                try
+                {
+                    _accessRightsUpdate.lock();
+
+                    _accessRights.remove(username);
+                    
+                    try
+                    {
+                        saveAccessFile();
+                    }
+                    catch (IOException e)
+                    {
+                        _logger.warn("Problem occured saving '" + _accessFile + "', the access right changes will not be preserved: " + e);
+                        return false;
+                    }
+                }
+                finally
+                {
+                    if (_accessRightsUpdate.isHeldByCurrentThread())
+                    {
+                        _accessRightsUpdate.unlock();
+                    }
+                }
+            }
+        }
+        catch (AccountNotFoundException e)
+        {
+            _logger.warn("Attempt to delete user (" + username + ") that doesn't exist");
+            return false;
+        }
+
+        return true;
+    }
+
+    public boolean reloadData()
+    {
+            try
+            {
+                loadAccessFile();
+                _principalDatabase.reload();
+            }
+            catch (ConfigurationException e)
+            {
+                _logger.warn("Reload failed due to:" + e);
+                return false;
+            }
+            catch (IOException e)
+            {
+                _logger.warn("Reload failed due to:" + e);
+                return false;
+            }
+            // Reload successful
+            return true;
+    }
+
+
+    @MBeanOperation(name = "viewUsers", description = "All users with access rights to the system.")
+    public TabularData viewUsers()
+    {
+        // Table of users
+        // Username(string), Access rights Read,Write,Admin(bool,bool,bool)
+
+        if (_userlistDataType == null)
+        {
+            _logger.warn("TabluarData not setup correctly");
+            return null;
+        }
+
+        List<Principal> users = _principalDatabase.getUsers();
+
+        TabularDataSupport userList = new TabularDataSupport(_userlistDataType);
+
+        try
+        {
+            // Create the tabular list of message header contents
+            for (Principal user : users)
+            {
+                // Create header attributes list
+
+                String rights = (String) _accessRights.get(user.getName());
+
+                Boolean read = false;
+                Boolean write = false;
+                Boolean admin = false;
+
+                if (rights != null)
+                {
+                    read = rights.equals(MBeanInvocationHandlerImpl.READONLY)
+                           || rights.equals(MBeanInvocationHandlerImpl.READWRITE);
+                    write = rights.equals(MBeanInvocationHandlerImpl.READWRITE);
+                    admin = rights.equals(MBeanInvocationHandlerImpl.ADMIN);
+                }
+
+                Object[] itemData = {user.getName(), read, write, admin};
+                CompositeData messageData = new CompositeDataSupport(_userDataType, _userItemNames, itemData);
+                userList.put(messageData);
+            }
+        }
+        catch (OpenDataException e)
+        {
+            _logger.warn("Unable to create user list due to :" + e);
+            return null;
+        }
+
+        return userList;
+    }
+
+    /*** Broker Methods **/
+
+    /**
+     * setPrincipalDatabase
+     *
+     * @param database set The Database to use for user lookup
+     */
+    public void setPrincipalDatabase(PrincipalDatabase database)
+    {
+        _principalDatabase = database;
+    }
+
+    /**
+     * setAccessFile
+     *
+     * @param accessFile the file to use for updating.
+     *
+     * @throws java.io.IOException If the file cannot be accessed
+     * @throws org.apache.commons.configuration.ConfigurationException
+     *                             if checks on the file fail.
+     */
+    public void setAccessFile(String accessFile) throws IOException, ConfigurationException
+    {
+        if (accessFile != null)
+        {
+            _accessFile = new File(accessFile);
+            if (!_accessFile.exists())
+            {
+                throw new ConfigurationException("'" + _accessFile + "' does not exist");
+            }
+
+            if (!_accessFile.canRead())
+            {
+                throw new ConfigurationException("Cannot read '" + _accessFile + "'.");
+            }
+
+            if (!_accessFile.canWrite())
+            {
+                _logger.warn("Unable to write to access rights file '" + _accessFile + "', changes will not be preserved.");
+            }
+
+            loadAccessFile();
+        }
+        else
+        {
+            _logger.warn("Access rights file specified is null. Access rights not changed.");
+        }
+    }
+
+    private void loadAccessFile() throws IOException, ConfigurationException
+    {
+        if(_accessFile == null)
+        {
+            _logger.error("No jmx access rights file has been specified.");
+            return;
+        }
+        
+        if(_accessFile.exists())
+        {
+            try
+            {
+                _accessRightsUpdate.lock();
+
+                Properties accessRights = new Properties();
+                accessRights.load(new FileInputStream(_accessFile));
+                checkAccessRights(accessRights);
+                setAccessRights(accessRights);
+            }
+            finally
+            {
+                if (_accessRightsUpdate.isHeldByCurrentThread())
+                {
+                    _accessRightsUpdate.unlock();
+                }
+            }
+        }
+        else
+        {
+            _logger.error("Specified jmxaccess rights file '" + _accessFile + "' does not exist.");
+        }
+    }
+
+    private void checkAccessRights(Properties accessRights)
+    {
+        Enumeration values = accessRights.propertyNames();
+
+        while (values.hasMoreElements())
+        {
+            String user = (String) values.nextElement();
+
+            if (_principalDatabase.getUser(user) == null)
+            {
+                _logger.warn("Access rights contains user '" + user + "' but there is no authentication data for that user");
+            }
+        }
+    }
+
+    private void saveAccessFile() throws IOException
+    {
+        try
+        {
+            _accessRightsUpdate.lock();
+
+            // Create temporary file
+            File tmp = File.createTempFile(_accessFile.getName(), ".tmp");
+
+            FileOutputStream output = new FileOutputStream(tmp);
+            _accessRights.store(output, "Generated by AMQUserManagementMBean Console : Last edited by user:" + getCurrentJMXUser());
+            output.close();
+
+            // Rename new file to main file
+            tmp.renameTo(_accessFile);
+
+            // delete tmp
+            tmp.delete();
+        }
+        finally
+        {
+            if (_accessRightsUpdate.isHeldByCurrentThread())
+            {
+                _accessRightsUpdate.unlock();
+            }
+        }
+        
+    }
+
+    private String getCurrentJMXUser()
+    {
+        AccessControlContext acc = AccessController.getContext();
+        
+        Subject subject = Subject.getSubject(acc);
+        if (subject == null)
+        {
+            return "Unknown user, authentication Subject was null";
+        }
+
+        // Retrieve JMXPrincipal from Subject
+        Set<JMXPrincipal> principals = subject.getPrincipals(JMXPrincipal.class);
+        if (principals == null || principals.isEmpty())
+        {
+            return "Unknown user principals were null";
+        }
+
+        Principal principal = principals.iterator().next();
+        return principal.getName();
+    }
+
+    /**
+     * user=read user=write user=readwrite user=admin
+     *
+     * @param accessRights The properties list of access rights to process
+     */
+    private void setAccessRights(Properties accessRights)
+    {
+        _logger.debug("Setting Access Rights:" + accessRights);
+        _accessRights = accessRights;
+        MBeanInvocationHandlerImpl.setAccessRights(_accessRights);
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/management/UserManagement.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/management/UserManagement.java
new file mode 100644
index 0000000000..9fcdd4cd17
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/management/UserManagement.java
@@ -0,0 +1,121 @@
+/*
+ *  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.management;
+
+import org.apache.qpid.server.management.MBeanOperation;
+import org.apache.qpid.server.management.MBeanOperationParameter;
+import org.apache.qpid.server.management.MBeanAttribute;
+import org.apache.qpid.AMQException;
+
+import javax.management.openmbean.TabularData;
+import javax.management.openmbean.CompositeData;
+import javax.management.JMException;
+import javax.management.MBeanOperationInfo;
+import java.io.IOException;
+
+public interface UserManagement
+{
+
+    String TYPE = "UserManagement";
+    int VERSION = 2;
+
+    //********** Operations *****************//
+    /**
+     * set password for user
+     *
+     * @param username The username to create
+     * @param password The password for the user
+     *
+     * @return The result of the operation
+     */
+    @MBeanOperation(name = "setPassword", description = "Set password for user.",
+                    impact = MBeanOperationInfo.ACTION)
+    boolean setPassword(@MBeanOperationParameter(name = "username", description = "Username")String username,
+                        @MBeanOperationParameter(name = "password", description = "Password")char[] password);
+
+    /**
+     * set rights for users with given details
+     *
+     * @param username The username to create
+     * @param read     The set of permission to give the new user
+     * @param write    The set of permission to give the new user
+     * @param admin    The set of permission to give the new user
+     *
+     * @return The result of the operation
+     */
+    @MBeanOperation(name = "setRights", description = "Set access rights for user.",
+                    impact = MBeanOperationInfo.ACTION)
+    boolean setRights(@MBeanOperationParameter(name = "username", description = "Username")String username,
+                      @MBeanOperationParameter(name = "read", description = "Administration read")boolean read,
+                      @MBeanOperationParameter(name = "readAndWrite", description = "Administration write")boolean write,
+                      @MBeanOperationParameter(name = "admin", description = "Administration rights")boolean admin);
+
+    /**
+     * Create users with given details
+     *
+     * @param username The username to create
+     * @param password The password for the user
+     * @param read     The set of permission to give the new user
+     * @param write    The set of permission to give the new user
+     * @param admin    The set of permission to give the new user
+     *
+     * @return The result of the operation
+     */
+    @MBeanOperation(name = "createUser", description = "Create new user from system.",
+                    impact = MBeanOperationInfo.ACTION)
+    boolean createUser(@MBeanOperationParameter(name = "username", description = "Username")String username,
+                       @MBeanOperationParameter(name = "password", description = "Password")char[] password,
+                       @MBeanOperationParameter(name = "read", description = "Administration read")boolean read,
+                       @MBeanOperationParameter(name = "readAndWrite", description = "Administration write")boolean write,
+                       @MBeanOperationParameter(name = "admin", description = "Administration rights")boolean admin);
+
+    /**
+     * View users returns all the users that are currently available to the system.
+     *
+     * @param username The user to delete
+     *
+     * @return The result of the operation
+     */
+    @MBeanOperation(name = "deleteUser", description = "Delete user from system.",
+                    impact = MBeanOperationInfo.ACTION)
+    boolean deleteUser(@MBeanOperationParameter(name = "username", description = "Username")String username);
+
+
+    /**
+     * Reload the date from disk
+     *
+     * @return The result of the operation
+     */
+    @MBeanOperation(name = "reloadData", description = "Reload the authentication file from disk.",
+                    impact = MBeanOperationInfo.ACTION)
+    boolean reloadData();
+
+    /**
+     * View users returns all the users that are currently available to the system.
+     *
+     * @return a table of users data (Username, read, write, admin)
+     */
+    @MBeanOperation(name = "viewUsers", description = "All users with access rights to the system.",
+                    impact = MBeanOperationInfo.INFO)
+    TabularData viewUsers();
+
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/AbstractACLPlugin.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/AbstractACLPlugin.java
new file mode 100644
index 0000000000..682135bc25
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/AbstractACLPlugin.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.access.plugins;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.security.access.ACLPlugin;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+
+/**
+ * This ACLPlugin abstains from all votes. Useful if your plugin only cares about a few operations. 
+ */
+public abstract class AbstractACLPlugin implements ACLPlugin
+{
+
+    private static final AuthzResult DEFAULT_ANSWER = AuthzResult.ABSTAIN;
+
+    public AuthzResult authoriseBind(AMQProtocolSession session, Exchange exch, AMQQueue queue,
+            AMQShortString routingKey)
+    {
+        return DEFAULT_ANSWER;
+    }
+
+    public AuthzResult authoriseConnect(AMQProtocolSession session, VirtualHost virtualHost)
+    {
+        return DEFAULT_ANSWER;
+    }
+
+    public AuthzResult authoriseConsume(AMQProtocolSession session, boolean noAck, AMQQueue queue)
+    {
+        return DEFAULT_ANSWER;
+    }
+
+    public AuthzResult authoriseConsume(AMQProtocolSession session, boolean exclusive, boolean noAck, boolean noLocal,
+            boolean nowait, AMQQueue queue)
+    {
+        return DEFAULT_ANSWER;
+    }
+
+    public AuthzResult authoriseCreateExchange(AMQProtocolSession session, boolean autoDelete, boolean durable,
+            AMQShortString exchangeName, boolean internal, boolean nowait, boolean passive, AMQShortString exchangeType)
+    {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
+    public AuthzResult authoriseCreateQueue(AMQProtocolSession session, boolean autoDelete, boolean durable,
+            boolean exclusive, boolean nowait, boolean passive, AMQShortString queue)
+    {
+        return DEFAULT_ANSWER;
+    }
+
+    public AuthzResult authoriseDelete(AMQProtocolSession session, AMQQueue queue)
+    {
+        return DEFAULT_ANSWER;
+    }
+
+    public AuthzResult authoriseDelete(AMQProtocolSession session, Exchange exchange)
+    {
+        return DEFAULT_ANSWER;
+    }
+
+    public AuthzResult authorisePublish(AMQProtocolSession session, boolean immediate, boolean mandatory,
+            AMQShortString routingKey, Exchange e)
+    {
+        return DEFAULT_ANSWER;
+    }
+
+    public AuthzResult authorisePurge(AMQProtocolSession session, AMQQueue queue)
+    {
+        return DEFAULT_ANSWER;
+    }
+
+    public AuthzResult authoriseUnbind(AMQProtocolSession session, Exchange exch, AMQShortString routingKey,
+            AMQQueue queue)
+    {
+        return DEFAULT_ANSWER;
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/AllowAll.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/AllowAll.java
new file mode 100644
index 0000000000..4af178574b
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/AllowAll.java
@@ -0,0 +1,54 @@
+/*
+ *  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 org.apache.commons.configuration.Configuration;
+import org.apache.qpid.server.security.access.ACLPlugin;
+import org.apache.qpid.server.security.access.ACLPluginFactory;
+
+public class AllowAll extends BasicACLPlugin
+{
+
+    public static final ACLPluginFactory FACTORY = new ACLPluginFactory()
+    {
+        public boolean supportsTag(String name)
+        {
+            return false;
+        }
+
+        public ACLPlugin newInstance(Configuration config)
+        {
+            return new AllowAll();
+        }
+    };
+
+    public String getPluginName()
+    {
+        return this.getClass().getSimpleName();
+    }
+
+    @Override
+    protected AuthzResult getResult()
+    {
+        // Always allow
+        return AuthzResult.ALLOWED;
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/BasicACLPlugin.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/BasicACLPlugin.java
new file mode 100644
index 0000000000..f7e537b02b
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/BasicACLPlugin.java
@@ -0,0 +1,129 @@
+/*
+ *  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 org.apache.commons.configuration.Configuration;
+import org.apache.qpid.AMQConnectionException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.security.access.ACLPlugin;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+
+public abstract class BasicACLPlugin implements ACLPlugin
+{
+
+    // Returns true or false if the plugin should authorise or deny the request    
+    protected abstract AuthzResult getResult();
+    
+    @Override
+    public AuthzResult authoriseBind(AMQProtocolSession session, Exchange exch,
+            AMQQueue queue, AMQShortString routingKey)
+    {
+        return getResult();
+    }
+
+    @Override
+    public AuthzResult authoriseConnect(AMQProtocolSession session,
+            VirtualHost virtualHost)
+    {
+        return getResult();
+    }
+
+    @Override
+    public AuthzResult authoriseConsume(AMQProtocolSession session, boolean noAck,
+            AMQQueue queue)
+    {
+        return getResult();    
+    }
+
+    @Override
+    public AuthzResult authoriseConsume(AMQProtocolSession session,
+            boolean exclusive, boolean noAck, boolean noLocal, boolean nowait,
+            AMQQueue queue)
+    {
+        return getResult();
+    }
+
+    @Override
+    public AuthzResult authoriseCreateExchange(AMQProtocolSession session,
+            boolean autoDelete, boolean durable, AMQShortString exchangeName,
+            boolean internal, boolean nowait, boolean passive,
+            AMQShortString exchangeType)
+    {
+        return getResult();
+    }
+
+    @Override
+    public AuthzResult authoriseCreateQueue(AMQProtocolSession session,
+            boolean autoDelete, boolean durable, boolean exclusive,
+            boolean nowait, boolean passive, AMQShortString queue)
+    {
+        return getResult();
+    }
+
+    @Override
+    public AuthzResult authoriseDelete(AMQProtocolSession session, AMQQueue queue)
+    {
+        return getResult();
+    }
+
+    @Override
+    public AuthzResult authoriseDelete(AMQProtocolSession session, Exchange exchange)
+    {
+        return getResult();
+    }
+
+    @Override
+    public AuthzResult authorisePublish(AMQProtocolSession session,
+            boolean immediate, boolean mandatory, AMQShortString routingKey,
+            Exchange e)
+    {
+        return getResult();
+    }
+
+    @Override
+    public AuthzResult authorisePurge(AMQProtocolSession session, AMQQueue queue)
+    {
+        return getResult();
+    }
+
+    @Override
+    public AuthzResult authoriseUnbind(AMQProtocolSession session, Exchange exch,
+            AMQShortString routingKey, AMQQueue queue)
+    {
+        return getResult();
+    }
+
+    @Override
+    public void setConfiguration(Configuration config)
+    {
+        // no-op
+    }
+
+    public boolean supportsTag(String name)
+    {
+        // This plugin doesn't support any tags
+        return false;
+    }
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/DenyAll.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/DenyAll.java
new file mode 100644
index 0000000000..26a76c9af1
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/DenyAll.java
@@ -0,0 +1,75 @@
+/*
+ *  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 org.apache.commons.configuration.Configuration;
+import org.apache.qpid.AMQConnectionException;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.protocol.AMQConstant;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.security.access.ACLManager;
+import org.apache.qpid.server.security.access.ACLPlugin;
+import org.apache.qpid.server.security.access.ACLPluginFactory;
+import org.apache.qpid.server.security.access.AccessResult;
+import org.apache.qpid.server.security.access.Permission;
+
+public class DenyAll extends BasicACLPlugin
+{
+    public static final ACLPluginFactory FACTORY = new ACLPluginFactory()
+    {
+        public boolean supportsTag(String name)
+        {
+            return false;
+        }
+
+        public ACLPlugin newInstance(Configuration config)
+        {
+            return new DenyAll();
+        }
+    };
+    
+    public AccessResult authorise(AMQProtocolSession session,
+            Permission permission, AMQMethodBody body, Object... parameters)
+            throws AMQConnectionException
+    {
+
+        if (ACLManager.getLogger().isInfoEnabled())
+        {
+            ACLManager.getLogger().info(
+                    "Denying user:" + session.getAuthorizedID());
+        }
+        throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
+                "DenyAll Plugin");
+    }
+
+    public String getPluginName()
+    {
+        return getClass().getSimpleName();
+    }
+
+    @Override 
+    protected AuthzResult getResult()
+    {
+        // Always deny
+        return AuthzResult.DENIED;
+    }
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/LegacyAccessPlugin.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/LegacyAccessPlugin.java
new file mode 100644
index 0000000000..fc1bc048d4
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/LegacyAccessPlugin.java
@@ -0,0 +1,71 @@
+/*
+ *  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.util.Collection;
+import java.util.HashSet;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.qpid.server.security.access.ACLPlugin;
+import org.apache.qpid.server.security.access.ACLPluginFactory;
+
+/**
+ * 
+ * Used to suppress warnings in legacy config files that have things in <security> which aren't handled by a plugin directly. 
+ *
+ */
+public class LegacyAccessPlugin extends BasicACLPlugin
+{
+    public static final ACLPluginFactory FACTORY = new ACLPluginFactory()
+    {
+        private Collection maskedTags = new HashSet<String>();
+        {
+            maskedTags.add("principal-databases");
+            maskedTags.add("access");
+            maskedTags.add("msg-auth");
+            maskedTags.add("false");
+            maskedTags.add("jmx");
+        }
+
+        public boolean supportsTag(String name)
+        {
+            return maskedTags .contains(name);
+        }
+
+        public ACLPlugin newInstance(Configuration config)
+        {
+            return new LegacyAccessPlugin();
+        }
+    };
+    
+    public String getPluginName()
+    {
+        return getClass().getSimpleName();
+    }
+
+    @Override 
+    protected AuthzResult getResult()
+    {
+        // Always abstain
+        return AuthzResult.ABSTAIN;
+    }
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/SimpleXML.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/SimpleXML.java
new file mode 100644
index 0000000000..2cc0c530de
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/SimpleXML.java
@@ -0,0 +1,432 @@
+/*
+ *  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 org.apache.commons.configuration.Configuration;
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQConnectionException;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.BasicConsumeBody;
+import org.apache.qpid.framing.BasicPublishBody;
+
+import org.apache.qpid.protocol.AMQConstant;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.security.access.ACLManager;
+import org.apache.qpid.server.security.access.ACLPlugin;
+import org.apache.qpid.server.security.access.ACLPluginFactory;
+import org.apache.qpid.server.security.access.AccessResult;
+import org.apache.qpid.server.security.access.Permission;
+import org.apache.qpid.server.security.access.PrincipalPermissions;
+import org.apache.qpid.server.security.access.ACLPlugin.AuthzResult;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * This uses the default
+ */
+public class SimpleXML implements ACLPlugin
+{
+    public static final ACLPluginFactory FACTORY = new ACLPluginFactory()
+    {
+        public boolean supportsTag(String name)
+        {
+            return name.startsWith("access_control_list");
+        }
+
+        public ACLPlugin newInstance(Configuration config)
+        {
+            SimpleXML plugin = new SimpleXML();
+            plugin.setConfiguration(config);
+            return plugin;
+        }
+    };
+    
+    private Map<String, PrincipalPermissions> _users;
+    private final AccessResult GRANTED = new AccessResult(this, AccessResult.AccessStatus.GRANTED);
+
+    public SimpleXML()
+    {
+        _users = new ConcurrentHashMap<String, PrincipalPermissions>();
+    }
+
+    public void setConfiguration(Configuration config)
+    {
+        processConfig(config);
+    }
+
+    private void processConfig(Configuration config)
+    {
+        processPublish(config);
+
+        processConsume(config);
+
+        processCreate(config);
+    }
+
+    /**
+     * Publish format takes Exchange + Routing Key Pairs
+     * 
+     * @param config
+     *            XML Configuration
+     */
+    private void processPublish(Configuration config)
+    {
+        Configuration publishConfig = config.subset("access_control_list.publish");
+
+        // Process users that have full publish permission
+        String[] users = publishConfig.getStringArray("users.user");
+
+        for (String user : users)
+        {
+            grant(Permission.PUBLISH, user);
+        }
+
+        // Process exchange limited users
+        int exchangeCount = 0;
+        Configuration exchangeConfig = publishConfig.subset("exchanges.exchange(" + exchangeCount + ")");
+
+        while (!exchangeConfig.isEmpty())
+        {
+            // Get Exchange Name
+            AMQShortString exchangeName = new AMQShortString(exchangeConfig.getString("name"));
+
+            // Get Routing Keys
+            int keyCount = 0;
+            Configuration routingkeyConfig = exchangeConfig.subset("routing_keys.routing_key(" + keyCount + ")");
+
+            while (!routingkeyConfig.isEmpty())
+            {
+                // Get RoutingKey Value
+                AMQShortString routingKeyValue = new AMQShortString(routingkeyConfig.getString("value"));
+
+                // Apply Exchange + RoutingKey permissions to Users
+                users = routingkeyConfig.getStringArray("users.user");
+                for (String user : users)
+                {
+                    grant(Permission.PUBLISH, user, exchangeName, routingKeyValue);
+                }
+
+                // Apply permissions to Groups
+
+                // Check for more configs
+                keyCount++;
+                routingkeyConfig = exchangeConfig.subset("routing_keys.routing_key(" + keyCount + ")");
+            }
+
+            // Apply Exchange wide permissions to Users
+            users = exchangeConfig.getStringArray("exchange(" + exchangeCount + ").users.user");
+
+            for (String user : users)
+            {
+                grant(Permission.PUBLISH, user, exchangeName);
+            }
+
+            // Apply permissions to Groups
+            exchangeCount++;
+            exchangeConfig = publishConfig.subset("exchanges.exchange(" + exchangeCount + ")");
+        }
+    }
+
+    private void grant(Permission permission, String user, Object... parameters)
+    {
+        PrincipalPermissions permissions = _users.get(user);
+
+        if (permissions == null)
+        {
+            permissions = new PrincipalPermissions(user);
+        }
+
+        _users.put(user, permissions);
+        permissions.grant(permission, parameters);
+    }
+
+    private void processConsume(Configuration config)
+    {
+        Configuration consumeConfig = config.subset("access_control_list.consume");
+
+        // Process queue limited users
+        int queueCount = 0;
+        Configuration queueConfig = consumeConfig.subset("queues.queue(" + queueCount + ")");
+
+        while (!queueConfig.isEmpty())
+        {
+            // Get queue Name
+            AMQShortString queueName = new AMQShortString(queueConfig.getString("name"));
+            // if there is no name then there may be a temporary element
+            boolean temporary = queueConfig.containsKey("temporary");
+            boolean ownQueues = queueConfig.containsKey("own_queues");
+
+            // Process permissions for this queue
+            String[] users = queueConfig.getStringArray("users.user");
+            for (String user : users)
+            {
+                grant(Permission.CONSUME, user, queueName, temporary, ownQueues);
+            }
+
+            // See if we have another config
+            queueCount++;
+            queueConfig = consumeConfig.subset("queues.queue(" + queueCount + ")");
+        }
+
+        // Process users that have full consume permission
+        String[] users = consumeConfig.getStringArray("users.user");
+
+        for (String user : users)
+        {
+            grant(Permission.CONSUME, user);
+        }
+    }
+
+    private void processCreate(Configuration config)
+    {
+        Configuration createConfig = config.subset("access_control_list.create");
+
+        // Process create permissions for queue creation
+        int queueCount = 0;
+        Configuration queueConfig = createConfig.subset("queues.queue(" + queueCount + ")");
+
+        while (!queueConfig.isEmpty())
+        {
+            // Get queue Name
+            AMQShortString queueName = new AMQShortString(queueConfig.getString("name"));
+
+            // if there is no name then there may be a temporary element
+            boolean temporary = queueConfig.containsKey("temporary");
+
+            int exchangeCount = 0;
+            Configuration exchangeConfig = queueConfig.subset("exchanges.exchange(" + exchangeCount + ")");
+
+            while (!exchangeConfig.isEmpty())
+            {
+
+                AMQShortString exchange = new AMQShortString(exchangeConfig.getString("name"));
+                AMQShortString routingKey = new AMQShortString(exchangeConfig.getString("routing_key"));
+
+                // Process permissions for this queue
+                String[] users = exchangeConfig.getStringArray("users.user");
+                for (String user : users)
+                {
+                    grant(Permission.CREATEEXCHANGE, user, exchange);
+                    grant(Permission.CREATEQUEUE, user, temporary, (queueName.equals("") ? null : queueName), (exchange
+                            .equals("") ? null : exchange), (routingKey.equals("") ? null : routingKey));
+                }
+
+                // See if we have another config
+                exchangeCount++;
+                exchangeConfig = queueConfig.subset("exchanges.exchange(" + exchangeCount + ")");
+            }
+
+            // Process users that are not bound to an exchange
+            String[] users = queueConfig.getStringArray("users.user");
+
+            for (String user : users)
+            {
+                grant(Permission.CREATEQUEUE, user, temporary, queueName);
+            }
+
+            // See if we have another config
+            queueCount++;
+            queueConfig = createConfig.subset("queues.queue(" + queueCount + ")");
+        }
+
+        // Process create permissions for exchange creation
+        int exchangeCount = 0;
+        Configuration exchangeConfig = createConfig.subset("exchanges.exchange(" + exchangeCount + ")");
+
+        while (!exchangeConfig.isEmpty())
+        {
+            AMQShortString exchange = new AMQShortString(exchangeConfig.getString("name"));
+            AMQShortString clazz = new AMQShortString(exchangeConfig.getString("class"));
+
+            // Process permissions for this queue
+            String[] users = exchangeConfig.getStringArray("users.user");
+            for (String user : users)
+            {
+                grant(Permission.CREATEEXCHANGE, user, exchange, clazz);
+            }
+
+            // See if we have another config
+            exchangeCount++;
+            exchangeConfig = queueConfig.subset("exchanges.exchange(" + exchangeCount + ")");
+        }
+
+        // Process users that have full create permission
+        String[] users = createConfig.getStringArray("users.user");
+
+        for (String user : users)
+        {
+            grant(Permission.CREATEEXCHANGE, user);
+            grant(Permission.CREATEQUEUE, user);
+        }
+
+    }
+
+    public String getPluginName()
+    {
+        return "Simple";
+    }
+
+    public AuthzResult authoriseBind(AMQProtocolSession session, Exchange exch, AMQQueue queue, AMQShortString routingKey)
+    {
+        PrincipalPermissions principalPermissions = _users.get(session.getAuthorizedID().getName());
+        if (principalPermissions == null)
+        {
+            return AuthzResult.DENIED;
+        }
+        else
+        {
+            return principalPermissions.authorise(Permission.BIND, null, exch, queue, routingKey);
+        }
+    }
+
+    public AuthzResult authoriseConnect(AMQProtocolSession session, VirtualHost virtualHost)
+    {
+        PrincipalPermissions principalPermissions = _users.get(session.getAuthorizedID().getName());
+        if (principalPermissions == null)
+        {
+            return AuthzResult.DENIED;
+        }
+        else
+        {
+            return principalPermissions.authorise(Permission.ACCESS);
+        }
+    }
+
+    public AuthzResult authoriseConsume(AMQProtocolSession session, boolean noAck, AMQQueue queue)
+    {
+        PrincipalPermissions principalPermissions = _users.get(session.getAuthorizedID().getName());
+        if (principalPermissions == null)
+        {
+            return AuthzResult.DENIED;
+        }
+        else
+        {
+            return principalPermissions.authorise(Permission.CONSUME, queue);
+        }
+    }
+
+    public AuthzResult authoriseConsume(AMQProtocolSession session, boolean exclusive, boolean noAck, boolean noLocal,
+            boolean nowait, AMQQueue queue)
+    {
+        return authoriseConsume(session, noAck, queue);
+    }
+
+    public AuthzResult authoriseCreateExchange(AMQProtocolSession session, boolean autoDelete, boolean durable,
+            AMQShortString exchangeName, boolean internal, boolean nowait, boolean passive, AMQShortString exchangeType)
+    {
+        PrincipalPermissions principalPermissions = _users.get(session.getAuthorizedID().getName());
+        if (principalPermissions == null)
+        {
+            return AuthzResult.DENIED;
+        }
+        else
+        {
+            return principalPermissions.authorise(Permission.CREATEEXCHANGE, exchangeName);
+        }
+    }
+
+    public AuthzResult authoriseCreateQueue(AMQProtocolSession session, boolean autoDelete, boolean durable, boolean exclusive,
+            boolean nowait, boolean passive, AMQShortString queue)
+    {
+        PrincipalPermissions principalPermissions = _users.get(session.getAuthorizedID().getName());
+        if (principalPermissions == null)
+        {
+            return AuthzResult.DENIED;
+        }
+        else
+        {
+            return principalPermissions.authorise(Permission.CREATEQUEUE, autoDelete, queue);
+        }
+    }
+
+    public AuthzResult authoriseDelete(AMQProtocolSession session, AMQQueue queue)
+    {
+        PrincipalPermissions principalPermissions = _users.get(session.getAuthorizedID().getName());
+        if (principalPermissions == null)
+        {
+            return AuthzResult.DENIED;
+        }
+        else
+        {
+            return principalPermissions.authorise(Permission.DELETE);
+        }
+    }
+
+    public AuthzResult authoriseDelete(AMQProtocolSession session, Exchange exchange)
+    {
+        PrincipalPermissions principalPermissions = _users.get(session.getAuthorizedID().getName());
+        if (principalPermissions == null)
+        {
+            return AuthzResult.DENIED;
+        }
+        else
+        {
+            return principalPermissions.authorise(Permission.DELETE);
+        }
+    }
+
+    public AuthzResult authorisePublish(AMQProtocolSession session, boolean immediate, boolean mandatory,
+            AMQShortString routingKey, Exchange e)
+    {
+        PrincipalPermissions principalPermissions = _users.get(session.getAuthorizedID().getName());
+        if (principalPermissions == null)
+        {
+            return AuthzResult.DENIED;
+        }
+        else
+        {
+            return principalPermissions.authorise(Permission.PUBLISH, e, routingKey);
+        }
+    }
+
+    public AuthzResult authorisePurge(AMQProtocolSession session, AMQQueue queue)
+    {
+        PrincipalPermissions principalPermissions = _users.get(session.getAuthorizedID().getName());
+        if (principalPermissions == null)
+        {
+            return AuthzResult.DENIED;
+        }
+        else
+        {
+            return principalPermissions.authorise(Permission.PURGE);
+        }
+    }
+
+    public AuthzResult authoriseUnbind(AMQProtocolSession session, Exchange exch, AMQShortString routingKey, AMQQueue queue)
+    {
+        PrincipalPermissions principalPermissions = _users.get(session.getAuthorizedID().getName());
+        if (principalPermissions == null)
+        {
+            return AuthzResult.DENIED;
+        }
+        else
+        {
+            return principalPermissions.authorise(Permission.UNBIND);
+        }
+    }
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/network/FirewallFactory.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/network/FirewallFactory.java
new file mode 100644
index 0000000000..a1a399e5bf
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/network/FirewallFactory.java
@@ -0,0 +1,45 @@
+/*
+ *
+ * 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.network;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.qpid.server.security.access.ACLPlugin;
+import org.apache.qpid.server.security.access.ACLPluginFactory;
+
+public class FirewallFactory implements ACLPluginFactory
+{
+
+    @Override
+    public ACLPlugin newInstance(Configuration config) throws ConfigurationException
+    {
+        FirewallPlugin plugin = new FirewallPlugin();
+        plugin.setConfiguration(config);
+        return plugin;
+    }
+
+    @Override
+    public boolean supportsTag(String name)
+    {
+        return name.equals("firewall");
+    }
+    
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/network/FirewallPlugin.java b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/network/FirewallPlugin.java
new file mode 100644
index 0000000000..85026121ab
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/access/plugins/network/FirewallPlugin.java
@@ -0,0 +1,264 @@
+/*
+ *
+ * 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.network;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Pattern;
+
+import org.apache.commons.configuration.CompositeConfiguration;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.XMLConfiguration;
+import org.apache.qpid.server.protocol.AMQMinaProtocolSession;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.security.access.ACLPlugin;
+import org.apache.qpid.server.security.access.ACLPluginFactory;
+import org.apache.qpid.server.security.access.plugins.AbstractACLPlugin;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.util.NetMatcher;
+
+public class FirewallPlugin extends AbstractACLPlugin
+{
+
+    public class FirewallPluginException extends Exception {}
+
+    public static final ACLPluginFactory FACTORY = new ACLPluginFactory()
+    {
+        public boolean supportsTag(String name)
+        {
+            return name.startsWith("firewall");
+        }
+
+        public ACLPlugin newInstance(Configuration config) throws ConfigurationException
+        {
+            FirewallPlugin plugin = new FirewallPlugin();
+            plugin.setConfiguration(config);
+            return plugin;
+        }
+    };
+    
+    public class FirewallRule
+    {
+
+        private static final long DNS_TIMEOUT = 30000;
+        private AuthzResult _access;
+        private NetMatcher _network;
+        private Pattern[] _hostnamePatterns;
+
+        public FirewallRule(String access, List networks, List hostnames)
+        {
+            _access = (access.equals("allow")) ? AuthzResult.ALLOWED : AuthzResult.DENIED;
+            
+            if (networks != null && networks.size() > 0)
+            {
+                String[] networkStrings = objListToStringArray(networks);
+                _network = new NetMatcher(networkStrings);
+            }
+            
+            if (hostnames != null && hostnames.size() > 0)
+            {
+                int i = 0;
+                _hostnamePatterns = new Pattern[hostnames.size()];
+                for (String hostname : objListToStringArray(hostnames))
+                {
+                    _hostnamePatterns[i++] = Pattern.compile(hostname);
+                }
+            }
+            
+        }
+
+        private String[] objListToStringArray(List objList)
+        {
+            String[] networkStrings = new String[objList.size()];
+            int i = 0;
+            for (Object network : objList)
+            {
+                networkStrings[i++] = (String) network;
+            }
+            return networkStrings;
+        }
+
+        public boolean match(InetAddress remote) throws FirewallPluginException
+        {
+            if (_hostnamePatterns != null)
+            {
+                String hostname = getHostname(remote);
+                if (hostname == null)
+                {
+                    throw new FirewallPluginException();
+                }
+                for (Pattern pattern : _hostnamePatterns)
+                {
+                    if (pattern.matcher(hostname).matches())
+                    {
+                        return true;
+                    }
+                }
+                return false;
+            }
+            else
+            {
+                return _network.matchInetNetwork(remote);
+            }
+        }
+
+        /**
+         * @param remote the InetAddress to look up
+         * @return the hostname, null if not found or takes longer than 30s to find
+         */
+        private String getHostname(final InetAddress remote)
+        {
+            final String[] hostname = new String[]{null};
+            final AtomicBoolean done = new AtomicBoolean(false);
+            // Spawn thread
+            Thread thread = new Thread(new Runnable()
+            {
+               public void run()
+               {
+                   hostname[0] = remote.getCanonicalHostName();
+                   done.getAndSet(true);
+                   synchronized (done)
+                   {
+                       done.notifyAll();
+                   }
+               }
+            });
+
+            thread.run();
+            long endTime = System.currentTimeMillis() + DNS_TIMEOUT;
+            
+            while (System.currentTimeMillis() < endTime && !done.get())
+            {
+                try
+                {
+                    synchronized (done)
+                    {
+                        done.wait(endTime - System.currentTimeMillis());
+                    }
+                }
+                catch (InterruptedException e)
+                {
+                    // Check the time and if necessary sleep for a bit longer
+                }
+            }
+            return hostname[0];
+        }
+
+        public AuthzResult getAccess()
+        {
+            return _access;
+        }
+
+    }
+
+    private AuthzResult _default = AuthzResult.ABSTAIN;
+    private FirewallRule[] _rules;
+
+    @Override
+    public AuthzResult authoriseConnect(AMQProtocolSession session, VirtualHost virtualHost)
+    {
+        if (!(session instanceof AMQMinaProtocolSession))
+        {
+            return AuthzResult.ABSTAIN; // We only deal with tcp sessions, which
+                                        // mean MINA right now
+        }
+
+        InetAddress addr = getInetAdressFromMinaSession((AMQMinaProtocolSession) session);
+
+        if (addr == null)
+        {
+            return AuthzResult.ABSTAIN; // Not an Inet socket on the other end
+        }
+
+        boolean match = false;
+        for (FirewallRule rule : _rules)
+        {
+            try
+            {
+                match = rule.match(addr);
+            }
+            catch (FirewallPluginException e)
+            {
+                return AuthzResult.DENIED;
+            }
+            if (match)
+            {
+                return rule.getAccess();
+            }
+        }
+        return _default;
+
+    }
+
+    private InetAddress getInetAdressFromMinaSession(AMQMinaProtocolSession session)
+    {
+        SocketAddress remote = session.getIOSession().getRemoteAddress();
+        if (remote instanceof InetSocketAddress)
+        {
+            return ((InetSocketAddress) remote).getAddress();
+        }
+        else
+        {
+            return null;
+        }
+    }
+
+    @Override
+    public void setConfiguration(Configuration config) throws ConfigurationException
+    {
+        // Get default action
+        String defaultAction = config.getString("[@default-action]");
+        if (defaultAction == null)
+        {
+            _default = AuthzResult.ABSTAIN;
+        }
+        else if (defaultAction.toLowerCase().equals("allow"))
+        {
+            _default = AuthzResult.ALLOWED;
+        }
+        else
+        {
+            _default = AuthzResult.DENIED;
+        }
+        CompositeConfiguration finalConfig = new CompositeConfiguration(config);
+        
+        List subFiles = config.getList("firewall.xml[@fileName]");
+        for (Object subFile : subFiles)
+        {
+            finalConfig.addConfiguration(new XMLConfiguration((String) subFile));
+        }
+
+        // all rules must have an access attribute
+        int numRules = finalConfig.getList("rule[@access]").size(); 
+        _rules = new FirewallRule[numRules];
+        for (int i = 0; i < numRules; i++)
+        {
+            FirewallRule rule = new FirewallRule(finalConfig.getString("rule(" + i + ")[@access]"), finalConfig.getList("rule("
+                    + i + ")[@network]"), finalConfig.getList("rule(" + i + ")[@hostname]"));
+            _rules[i] = rule;
+        }
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/AuthenticationResult.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/AuthenticationResult.java
new file mode 100644
index 0000000000..3f846b9dd0
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/AuthenticationResult.java
@@ -0,0 +1,63 @@
+/*
+ *
+ * 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;
+
+import javax.security.sasl.SaslException;
+
+public class AuthenticationResult
+{
+    public enum AuthenticationStatus
+    {
+        SUCCESS, CONTINUE, ERROR
+    }
+
+    public AuthenticationStatus status;
+    public byte[] challenge;
+    
+    private Exception cause;
+
+    public AuthenticationResult(AuthenticationStatus status)
+    {
+        this(null, status, null);
+    }
+
+    public AuthenticationResult(byte[] challenge, AuthenticationStatus status)
+    {
+        this(challenge, status, null);
+    }
+
+    public AuthenticationResult(AuthenticationStatus error, Exception cause)
+    {
+        this(null, error, cause);
+    }
+
+    public AuthenticationResult(byte[] challenge, AuthenticationStatus status, Exception cause)
+    {
+        this.status = status;
+        this.challenge = challenge;
+        this.cause = cause;
+    }
+
+    public Exception getCause()
+    {
+        return cause;
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/Base64MD5PasswordFilePrincipalDatabase.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/Base64MD5PasswordFilePrincipalDatabase.java
new file mode 100644
index 0000000000..3c211746e3
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/Base64MD5PasswordFilePrincipalDatabase.java
@@ -0,0 +1,541 @@
+/*
+ *  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.database;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.server.security.access.management.AMQUserManagementMBean;
+import org.apache.qpid.server.security.auth.sasl.AuthenticationProviderInitialiser;
+import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
+import org.apache.qpid.server.security.auth.sasl.crammd5.CRAMMD5HashedInitialiser;
+
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.login.AccountNotFoundException;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.security.Principal;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.regex.Pattern;
+
+/**
+ * Represents a user database where the account information is stored in a simple flat file.
+ *
+ * The file is expected to be in the form: username:password username1:password1 ... usernamen:passwordn
+ *
+ * where a carriage return separates each username/password pair. Passwords are assumed to be in plain text.
+ */
+public class Base64MD5PasswordFilePrincipalDatabase implements PrincipalDatabase
+{
+    private static final Logger _logger = Logger.getLogger(Base64MD5PasswordFilePrincipalDatabase.class);
+
+    private File _passwordFile;
+
+    private Pattern _regexp = Pattern.compile(":");
+
+    private Map<String, AuthenticationProviderInitialiser> _saslServers;
+
+    AMQUserManagementMBean _mbean;
+    public static final String DEFAULT_ENCODING = "utf-8";
+    private Map<String, HashedUser> _users = new HashMap<String, HashedUser>();
+    private ReentrantLock _userUpdate = new ReentrantLock();
+
+    public Base64MD5PasswordFilePrincipalDatabase()
+    {
+        _saslServers = new HashMap<String, AuthenticationProviderInitialiser>();
+
+        /**
+         *  Create Authenticators for MD5 Password file.
+         */
+
+        // Accept Plain incomming and hash it for comparison to the file.
+        CRAMMD5HashedInitialiser cram = new CRAMMD5HashedInitialiser();
+        cram.initialise(this);
+        _saslServers.put(cram.getMechanismName(), cram);
+
+        //fixme The PDs should setup a PD Mangement MBean
+//        try
+//        {
+//            _mbean = new AMQUserManagementMBean();
+//            _mbean.setPrincipalDatabase(this);
+//        }
+//        catch (JMException e)
+//        {
+//            _logger.warn("User management disabled as unable to create MBean:" + e);
+//        }
+    }
+
+    public void setPasswordFile(String passwordFile) throws IOException
+    {
+        File f = new File(passwordFile);
+        _logger.info("PasswordFilePrincipalDatabase using file " + f.getAbsolutePath());
+        _passwordFile = f;
+        if (!f.exists())
+        {
+            throw new FileNotFoundException("Cannot find password file " + f);
+        }
+        if (!f.canRead())
+        {
+            throw new FileNotFoundException("Cannot read password file " + f +
+                                            ". Check permissions.");
+        }
+
+        loadPasswordFile();
+    }
+
+    /**
+     * SASL Callback Mechanism - sets the Password in the PasswordCallback based on the value in the PasswordFile
+     * If you want to change the password for a user, use updatePassword instead.
+     *
+     * @param principal The Principal to set the password for
+     * @param callback  The PasswordCallback to call setPassword on
+     *
+     * @throws AccountNotFoundException If the Principal cannont be found in this Database
+     */
+    public void setPassword(Principal principal, PasswordCallback callback) throws AccountNotFoundException
+    {
+        if (_passwordFile == null)
+        {
+            throw new AccountNotFoundException("Unable to locate principal since no password file was specified during initialisation");
+        }
+        if (principal == null)
+        {
+            throw new IllegalArgumentException("principal must not be null");
+        }
+
+        char[] pwd = lookupPassword(principal.getName());
+
+        if (pwd != null)
+        {
+            callback.setPassword(pwd);
+        }
+        else
+        {
+            throw new AccountNotFoundException("No account found for principal " + principal);
+        }
+    }
+
+    /**
+     * Used to verify that the presented Password is correct. Currently only used by Management Console
+     *
+     * @param principal The principal to authenticate
+     * @param password  The password to check
+     *
+     * @return true if password is correct
+     *
+     * @throws AccountNotFoundException if the principal cannot be found
+     */
+    public boolean verifyPassword(String principal, char[] password) throws AccountNotFoundException
+    {
+        char[] pwd = lookupPassword(principal);
+        
+        if (pwd == null)
+        {
+            throw new AccountNotFoundException("Unable to lookup the specfied users password");
+        }
+        
+        byte[] byteArray = new byte[password.length];
+        int index = 0;
+        for (char c : password)
+        {
+            byteArray[index++] = (byte) c;
+        }
+        
+        byte[] MD5byteArray;
+        try
+        {
+            MD5byteArray = HashedUser.getMD5(byteArray);
+        }
+        catch (Exception e1)
+        {
+            _logger.warn("Unable to hash password for user '" + principal + "' for comparison");
+            return false;
+        }
+        
+        char[] hashedPassword = new char[MD5byteArray.length];
+
+        index = 0;
+        for (byte c : MD5byteArray)
+        {
+            hashedPassword[index++] = (char) c;
+        }
+
+        return compareCharArray(pwd, hashedPassword);
+    }
+    
+    private boolean compareCharArray(char[] a, char[] b)
+    {
+        boolean equal = false;
+        if (a.length == b.length)
+        {
+            equal = true;
+            int index = 0;
+            while (equal && index < a.length)
+            {
+                equal = a[index] == b[index];
+                index++;
+            }
+        }
+        return equal;
+    }
+
+    /**
+     * Changes the password for the specified user
+     * 
+     * @param principal to change the password for
+     * @param password plaintext password to set the password too
+     */
+    public boolean updatePassword(Principal principal, char[] password) throws AccountNotFoundException
+    {
+        HashedUser user = _users.get(principal.getName());
+
+        if (user == null)
+        {
+            throw new AccountNotFoundException(principal.getName());
+        }
+
+        try
+        {
+            try
+            {
+                _userUpdate.lock();
+                char[] orig = user.getPassword();
+                user.setPassword(password,false);
+
+                try
+                {
+                    savePasswordFile();
+                }
+                catch (IOException e)
+                {
+                    _logger.error("Unable to save password file, password change for user'"
+                                  + principal + "' will revert at restart");
+                    //revert the password change
+                    user.setPassword(orig,true);
+                    return false;
+                }
+                return true;
+            }
+            finally
+            {
+                if (_userUpdate.isHeldByCurrentThread())
+                {
+                    _userUpdate.unlock();
+                }
+            }
+        }
+        catch (Exception e)
+        {
+            return false;
+        }
+    }
+
+    public boolean createPrincipal(Principal principal, char[] password)
+    {
+        if (_users.get(principal.getName()) != null)
+        {
+            return false;
+        }
+
+        HashedUser user;
+        try
+        {
+            user = new HashedUser(principal.getName(), password);
+        }
+        catch (Exception e1)
+        {
+            _logger.warn("Unable to create new user '" + principal.getName() + "'");
+            return false;
+        }
+
+
+        try
+        {
+            _userUpdate.lock();
+            _users.put(user.getName(), user);
+
+            try
+            {
+                savePasswordFile();
+                return true;
+            }
+            catch (IOException e)
+            {
+                //remove the use on failure.
+                _users.remove(user.getName());
+                return false;
+            }
+        }
+        finally
+        {
+            if (_userUpdate.isHeldByCurrentThread())
+            {
+                _userUpdate.unlock();
+            }
+        }
+    }
+
+    public boolean deletePrincipal(Principal principal) throws AccountNotFoundException
+    {
+        HashedUser user = _users.get(principal.getName());
+
+        if (user == null)
+        {
+            throw new AccountNotFoundException(principal.getName());
+        }
+
+        try
+        {
+            _userUpdate.lock();
+            user.delete();
+
+            try
+            {
+                savePasswordFile();
+            }
+            catch (IOException e)
+            {
+                _logger.warn("Unable to remove user '" + user.getName() + "' from password file.");
+                return false;
+            }
+
+            _users.remove(user.getName());
+        }
+        finally
+        {
+            if (_userUpdate.isHeldByCurrentThread())
+            {
+                _userUpdate.unlock();
+            }
+        }
+
+        return true;
+    }
+
+    public Map<String, AuthenticationProviderInitialiser> getMechanisms()
+    {
+        return _saslServers;
+    }
+
+    public List<Principal> getUsers()
+    {
+        return new LinkedList<Principal>(_users.values());
+    }
+
+    public Principal getUser(String username)
+    {
+        if (_users.containsKey(username))
+        {
+            return new UsernamePrincipal(username);
+        }
+        return null;
+    }
+
+    /**
+     * Looks up the password for a specified user in the password file. Note this code is <b>not</b> secure since it
+     * creates strings of passwords. It should be modified to create only char arrays which get nulled out.
+     *
+     * @param name The principal name to lookup
+     *
+     * @return a char[] for use in SASL.
+     */
+    private char[] lookupPassword(String name)
+    {
+        HashedUser user = _users.get(name);
+        if (user == null)
+        {
+            return null;
+        }
+        else
+        {
+            return user.getPassword();
+        }
+    }
+
+    private void loadPasswordFile() throws IOException
+    {
+        try
+        {
+            _userUpdate.lock();
+            _users.clear();
+
+            BufferedReader reader = null;
+            try
+            {
+                reader = new BufferedReader(new FileReader(_passwordFile));
+                String line;
+
+                while ((line = reader.readLine()) != null)
+                {
+                    String[] result = _regexp.split(line);
+                    if (result == null || result.length < 2 || result[0].startsWith("#"))
+                    {
+                        continue;
+                    }
+
+                    HashedUser user = new HashedUser(result);
+                    _logger.info("Created user:" + user);
+                    _users.put(user.getName(), user);
+                }
+            }
+            finally
+            {
+                if (reader != null)
+                {
+                    reader.close();
+                }
+            }
+        }
+        finally
+        {
+            if (_userUpdate.isHeldByCurrentThread())
+            {
+                _userUpdate.unlock();
+            }
+        }
+    }
+
+    private void savePasswordFile() throws IOException
+    {
+        try
+        {
+            _userUpdate.lock();
+
+            BufferedReader reader = null;
+            PrintStream writer = null;
+            File tmp = File.createTempFile(_passwordFile.getName(), ".tmp");
+
+            try
+            {
+                writer = new PrintStream(tmp);
+                reader = new BufferedReader(new FileReader(_passwordFile));
+                String line;
+
+                while ((line = reader.readLine()) != null)
+                {
+                    String[] result = _regexp.split(line);
+                    if (result == null || result.length < 2 || result[0].startsWith("#"))
+                    {
+                        writer.write(line.getBytes(DEFAULT_ENCODING));
+                        writer.println();
+                        continue;
+                    }
+
+                    HashedUser user = _users.get(result[0]);
+
+                    if (user == null)
+                    {
+                        writer.write(line.getBytes(DEFAULT_ENCODING));
+                        writer.println();
+                    }
+                    else if (!user.isDeleted())
+                    {
+                        if (!user.isModified())
+                        {
+                            writer.write(line.getBytes(DEFAULT_ENCODING));
+                            writer.println();
+                        }
+                        else
+                        {
+                            try
+                            {
+                                byte[] encodedPassword = user.getEncodedPassword();
+
+                                writer.write((user.getName() + ":").getBytes(DEFAULT_ENCODING));
+                                writer.write(encodedPassword);
+                                writer.println();
+
+                                user.saved();
+                            }
+                            catch (Exception e)
+                            {
+                                _logger.warn("Unable to encode new password reverting to old password.");
+                                writer.write(line.getBytes(DEFAULT_ENCODING));
+                                writer.println();
+                            }
+                        }
+                    }
+                }
+
+                for (HashedUser user : _users.values())
+                {
+                    if (user.isModified())
+                    {
+                        byte[] encodedPassword;
+                        try
+                        {
+                            encodedPassword = user.getEncodedPassword();
+                            writer.write((user.getName() + ":").getBytes(DEFAULT_ENCODING));
+                            writer.write(encodedPassword);
+                            writer.println();
+                            user.saved();
+                        }
+                        catch (Exception e)
+                        {
+                            _logger.warn("Unable to get Encoded password for user'" + user.getName() + "' password not saved");
+                        }
+                    }
+                }
+            }
+            finally
+            {
+                if (reader != null)
+                {
+                    reader.close();
+                }
+
+                if (writer != null)
+                {
+                    writer.close();
+                }
+
+                // Swap temp file to main password file.
+                File old = new File(_passwordFile.getAbsoluteFile() + ".old");
+                if (old.exists())
+                {
+                    old.delete();
+                }
+                _passwordFile.renameTo(old);
+                tmp.renameTo(_passwordFile);
+                tmp.delete();
+            }
+        }
+        finally
+        {
+            if (_userUpdate.isHeldByCurrentThread())
+            {
+                _userUpdate.unlock();
+            }
+        }
+    }
+
+    public void reload() throws IOException
+    {
+        loadPasswordFile();
+    }
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/ConfigurationFilePrincipalDatabaseManager.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/ConfigurationFilePrincipalDatabaseManager.java
new file mode 100644
index 0000000000..e0d4c49af1
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/ConfigurationFilePrincipalDatabaseManager.java
@@ -0,0 +1,232 @@
+/*
+ *
+ * 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.database;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.configuration.PropertyUtils;
+import org.apache.qpid.configuration.PropertyException;
+import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.security.auth.database.PrincipalDatabase;
+import org.apache.qpid.server.security.auth.database.PrincipalDatabaseManager;
+import org.apache.qpid.server.security.access.management.AMQUserManagementMBean;
+import org.apache.qpid.AMQException;
+
+import javax.management.JMException;
+
+public class ConfigurationFilePrincipalDatabaseManager implements PrincipalDatabaseManager
+{
+    private static final Logger _logger = Logger.getLogger(ConfigurationFilePrincipalDatabaseManager.class);
+
+    Map<String, PrincipalDatabase> _databases;
+
+    public ConfigurationFilePrincipalDatabaseManager(ServerConfiguration _configuration) throws Exception
+    {
+        _logger.info("Initialising PrincipleDatabase authentication manager");
+        _databases = initialisePrincipalDatabases(_configuration);
+    }
+
+    private Map<String, PrincipalDatabase> initialisePrincipalDatabases(ServerConfiguration _configuration) throws Exception
+    {
+        List<String> databaseNames = _configuration.getPrincipalDatabaseNames();
+        List<String> databaseClasses = _configuration.getPrincipalDatabaseClass();
+        Map<String, PrincipalDatabase> databases = new HashMap<String, PrincipalDatabase>();
+
+        if (databaseNames.size() == 0)
+        {
+            _logger.warn("No Principal databases specified. Broker running with NO AUTHENTICATION");
+        }
+
+        for (int i = 0; i < databaseNames.size(); i++)
+        {
+            Object o;
+            try
+            {
+                o = Class.forName(databaseClasses.get(i)).newInstance();
+            }
+            catch (Exception e)
+            {
+                throw new Exception("Error initialising principal database: " + e, e);
+            }
+
+            if (!(o instanceof PrincipalDatabase))
+            {
+                throw new Exception("Principal databases must implement the PrincipalDatabase interface");
+            }
+
+            initialisePrincipalDatabase((PrincipalDatabase) o, _configuration, i);
+
+            String name = databaseNames.get(i);
+            if ((name == null) || (name.length() == 0))
+            {
+                throw new Exception("Principal database names must have length greater than or equal to one character");
+            }
+
+            PrincipalDatabase pd = databases.get(name);
+            if (pd != null)
+            {
+                throw new Exception("Duplicate principal database name not permitted");
+            }
+
+            _logger.info("Initialised principal database '" + name + "' successfully");
+            databases.put(name, (PrincipalDatabase) o);
+        }
+
+        return databases;
+    }
+
+    private void initialisePrincipalDatabase(PrincipalDatabase principalDatabase, ServerConfiguration _configuration, int index)
+            throws FileNotFoundException, ConfigurationException
+    {
+        List<String> argumentNames = _configuration.getPrincipalDatabaseAttributeNames(index);
+        List<String> argumentValues = _configuration.getPrincipalDatabaseAttributeValues(index);
+        for (int i = 0; i < argumentNames.size(); i++)
+        {
+            String argName = argumentNames.get(i);
+            if ((argName == null) || (argName.length() == 0))
+            {
+                throw new ConfigurationException("Argument names must have length >= 1 character");
+            }
+
+            if (Character.isLowerCase(argName.charAt(0)))
+            {
+                argName = Character.toUpperCase(argName.charAt(0)) + argName.substring(1);
+            }
+
+            String methodName = "set" + argName;
+            Method method = null;
+            try
+            {
+                method = principalDatabase.getClass().getMethod(methodName, String.class);
+            }
+            catch (Exception e)
+            {
+                // do nothing.. as on error method will be null
+            }
+
+            if (method == null)
+            {
+                throw new ConfigurationException("No method " + methodName + " found in class "
+                                                 + principalDatabase.getClass()
+                                                 + " hence unable to configure principal database. The method must be public and "
+                                                 + "have a single String argument with a void return type");
+            }
+
+            try
+            {
+                method.invoke(principalDatabase, PropertyUtils.replaceProperties(argumentValues.get(i)));
+            }
+            catch (Exception ite)
+            {
+                if (ite instanceof ConfigurationException)
+                {
+                    throw(ConfigurationException) ite;
+                }
+                else
+                {
+                    throw new ConfigurationException(ite.getMessage(), ite);
+                }
+            }
+        }
+    }
+
+    public Map<String, PrincipalDatabase> getDatabases()
+    {
+        return _databases;
+    }
+
+    public void initialiseManagement(ServerConfiguration config) throws ConfigurationException
+    {
+        try
+        {
+            AMQUserManagementMBean _mbean = new AMQUserManagementMBean();
+
+            List<String> principalDBs = config.getManagementPrincipalDBs();
+
+            if (principalDBs.size() == 0)
+            {
+                throw new ConfigurationException("No principal-database specified for jmx security");
+            }
+
+            String databaseName = principalDBs.get(0);
+
+            PrincipalDatabase database = getDatabases().get(databaseName);
+
+            if (database == null)
+            {
+                throw new ConfigurationException("Principal-database '" + databaseName + "' not found");
+            }
+
+            _mbean.setPrincipalDatabase(database);
+
+            List<String> jmxaccesslist = config.getManagementAccessList();
+
+            if (jmxaccesslist.size() == 0)
+            {
+                throw new ConfigurationException("No access control files specified for jmx security");
+            }
+
+            String jmxaccesssFile = null;
+
+            try
+            {
+                jmxaccesssFile = PropertyUtils.replaceProperties(jmxaccesslist.get(0));
+            }
+            catch (PropertyException e)
+            {
+                throw new ConfigurationException("Unable to parse access control filename '" + jmxaccesssFile + "'");
+            }
+
+            try
+            {
+                _mbean.setAccessFile(jmxaccesssFile);
+            }
+            catch (IOException e)
+            {
+                _logger.warn("Unable to load access file:" + jmxaccesssFile);
+            }
+
+            try
+            {
+                _mbean.register();
+            }
+            catch (AMQException e)
+            {
+                _logger.warn("Unable to register user management MBean");
+            }
+        }
+        catch (JMException e)
+        {
+            _logger.warn("User management disabled as unable to create MBean:" + e);
+        }
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/HashedUser.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/HashedUser.java
new file mode 100644
index 0000000000..3690e7f92a
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/HashedUser.java
@@ -0,0 +1,169 @@
+/*
+*
+ * 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.database;
+
+import org.apache.commons.codec.EncoderException;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.log4j.Logger;
+
+import java.io.UnsupportedEncodingException;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.security.Principal;
+
+public class HashedUser implements Principal
+{
+    private static final Logger _logger = Logger.getLogger(HashedUser.class);
+
+    String _name;
+    char[] _password;
+    byte[] _encodedPassword = null;
+    private boolean _modified = false;
+    private boolean _deleted = false;
+
+    HashedUser(String[] data) throws UnsupportedEncodingException
+    {
+        if (data.length != 2)
+        {
+            throw new IllegalArgumentException("User Data should be length 2, username, password");
+        }
+
+        _name = data[0];
+
+        byte[] encoded_password = data[1].getBytes(Base64MD5PasswordFilePrincipalDatabase.DEFAULT_ENCODING);
+
+        Base64 b64 = new Base64();
+        byte[] decoded = b64.decode(encoded_password);
+
+        _encodedPassword = encoded_password;
+
+        _password = new char[decoded.length];
+
+        int index = 0;
+        for (byte c : decoded)
+        {
+            _password[index++] = (char) c;
+        }
+    }
+
+    public HashedUser(String name, char[] password) throws UnsupportedEncodingException, NoSuchAlgorithmException
+    {
+        _name = name;
+        setPassword(password,false);
+    }
+
+    public static byte[] getMD5(byte[] data) throws NoSuchAlgorithmException, UnsupportedEncodingException
+    {
+        MessageDigest md = MessageDigest.getInstance("MD5");
+
+        for (byte b : data)
+        {
+            md.update(b);
+        }
+
+        return md.digest();
+    }
+
+    public String getName()
+    {
+        return _name;
+    }
+
+    public String toString()
+    {
+        return _name;
+    }
+
+    char[] getPassword()
+    {
+        return _password;
+    }
+
+    void setPassword(char[] password, boolean alreadyHashed) throws UnsupportedEncodingException,  NoSuchAlgorithmException
+    {
+        if(alreadyHashed){
+            _password = password;
+        }
+        else
+        {
+            byte[] byteArray = new byte[password.length];
+            int index = 0;
+            for (char c : password)
+            {
+                byteArray[index++] = (byte) c;
+            }
+            
+            byte[] MD5byteArray = getMD5(byteArray);
+            
+            _password = new char[MD5byteArray.length];
+
+            index = 0;
+            for (byte c : MD5byteArray)
+            {
+                _password[index++] = (char) c;
+            }
+        }
+        
+        _modified = true;
+        _encodedPassword = null;
+    }
+
+    byte[] getEncodedPassword() throws EncoderException, UnsupportedEncodingException, NoSuchAlgorithmException
+    {
+        if (_encodedPassword == null)
+        {
+            encodePassword();
+        }
+        return _encodedPassword;
+    }
+
+    private void encodePassword() throws EncoderException, UnsupportedEncodingException, NoSuchAlgorithmException
+    {
+        byte[] byteArray = new byte[_password.length];
+        int index = 0;
+        for (char c : _password)
+        {
+            byteArray[index++] = (byte) c;
+        }
+        _encodedPassword = (new Base64()).encode(byteArray);
+    }
+
+    public boolean isModified()
+    {
+        return _modified;
+    }
+
+    public boolean isDeleted()
+    {
+        return _deleted;
+    }
+
+    public void delete()
+    {
+        _deleted = true;
+    }
+
+    public void saved()
+    {
+        _modified = false;
+    }
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PlainPasswordFilePrincipalDatabase.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PlainPasswordFilePrincipalDatabase.java
new file mode 100644
index 0000000000..5e4678a63b
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PlainPasswordFilePrincipalDatabase.java
@@ -0,0 +1,491 @@
+/*
+ *  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.database;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.server.security.auth.sasl.AuthenticationProviderInitialiser;
+import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
+import org.apache.qpid.server.security.auth.sasl.amqplain.AmqPlainInitialiser;
+import org.apache.qpid.server.security.auth.sasl.crammd5.CRAMMD5Initialiser;
+import org.apache.qpid.server.security.auth.sasl.plain.PlainInitialiser;
+
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.login.AccountNotFoundException;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.security.Principal;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.regex.Pattern;
+
+/**
+ * Represents a user database where the account information is stored in a simple flat file.
+ *
+ * The file is expected to be in the form: username:password username1:password1 ... usernamen:passwordn
+ *
+ * where a carriage return separates each username/password pair. Passwords are assumed to be in plain text.
+ */
+public class PlainPasswordFilePrincipalDatabase implements PrincipalDatabase
+{
+    public static final String DEFAULT_ENCODING = "utf-8";
+    
+    private static final Logger _logger = Logger.getLogger(PlainPasswordFilePrincipalDatabase.class);
+
+    private File _passwordFile;
+
+    private Pattern _regexp = Pattern.compile(":");
+
+    private Map<String, AuthenticationProviderInitialiser> _saslServers;    
+    
+    private Map<String, PlainUser> _users = new HashMap<String, PlainUser>();
+    private ReentrantLock _userUpdate = new ReentrantLock();
+
+    public PlainPasswordFilePrincipalDatabase()
+    {
+        _saslServers = new HashMap<String, AuthenticationProviderInitialiser>();
+
+        /**
+         *  Create Authenticators for Plain Password file.
+         */
+
+        // Accept AMQPlain incomming and compare it to the file.
+        AmqPlainInitialiser amqplain = new AmqPlainInitialiser();
+        amqplain.initialise(this);
+
+        // Accept Plain incomming and compare it to the file.
+        PlainInitialiser plain = new PlainInitialiser();
+        plain.initialise(this);
+
+        //  Accept MD5 incomming and Hash file value for comparison
+        CRAMMD5Initialiser cram = new CRAMMD5Initialiser();
+        cram.initialise(this);
+
+        _saslServers.put(amqplain.getMechanismName(), amqplain);
+        _saslServers.put(plain.getMechanismName(), plain);
+        _saslServers.put(cram.getMechanismName(), cram);
+    }
+
+    public void setPasswordFile(String passwordFile) throws IOException
+    {
+        File f = new File(passwordFile);
+        _logger.info("PlainPasswordFile using file " + f.getAbsolutePath());
+        _passwordFile = f;
+        if (!f.exists())
+        {
+            throw new FileNotFoundException("Cannot find password file " + f);
+        }
+        if (!f.canRead())
+        {
+            throw new FileNotFoundException("Cannot read password file " + f +
+                                            ". Check permissions.");
+        }
+        
+        loadPasswordFile();
+    }
+
+    /**
+     * SASL Callback Mechanism - sets the Password in the PasswordCallback based on the value in the PasswordFile
+     * If you want to change the password for a user, use updatePassword instead.
+     *
+     * @param principal The Principal to set the password for
+     * @param callback  The PasswordCallback to call setPassword on
+     *
+     * @throws AccountNotFoundException If the Principal cannot be found in this Database
+     */
+    public void setPassword(Principal principal, PasswordCallback callback) throws AccountNotFoundException
+    {
+        if (_passwordFile == null)
+        {
+            throw new AccountNotFoundException("Unable to locate principal since no password file was specified during initialisation");
+        }
+        if (principal == null)
+        {
+            throw new IllegalArgumentException("principal must not be null");
+        }
+        char[] pwd = lookupPassword(principal.getName());
+        
+        if (pwd != null)
+        {
+            callback.setPassword(pwd);
+        }
+        else
+        {
+            throw new AccountNotFoundException("No account found for principal " + principal);
+        }
+    }
+
+    /**
+     * Used to verify that the presented Password is correct. Currently only used by Management Console
+     *
+     * @param principal The principal to authenticate
+     * @param password  The plaintext password to check
+     *
+     * @return true if password is correct
+     *
+     * @throws AccountNotFoundException if the principal cannot be found
+     */
+    public boolean verifyPassword(String principal, char[] password) throws AccountNotFoundException
+    {
+
+        char[] pwd = lookupPassword(principal);
+        
+        if (pwd == null)
+        {
+            throw new AccountNotFoundException("Unable to lookup the specfied users password");
+        }
+
+        return compareCharArray(pwd, password);
+
+    }
+
+    /**
+     * Changes the password for the specified user
+     * 
+     * @param principal to change the password for
+     * @param password plaintext password to set the password too
+     */
+    public boolean updatePassword(Principal principal, char[] password) throws AccountNotFoundException
+    {
+        PlainUser user = _users.get(principal.getName());
+
+        if (user == null)
+        {
+            throw new AccountNotFoundException(principal.getName());
+        }
+
+        try
+        {
+            try
+            {
+                _userUpdate.lock();
+                char[] orig = user.getPassword();
+                user.setPassword(password);
+
+                try
+                {
+                    savePasswordFile();
+                }
+                catch (IOException e)
+                {
+                    _logger.error("Unable to save password file, password change for user '" + principal + "' discarded");
+                    //revert the password change
+                    user.setPassword(orig);
+                    return false;
+                }
+                return true;
+            }
+            finally
+            {
+                if (_userUpdate.isHeldByCurrentThread())
+                {
+                    _userUpdate.unlock();
+                }
+            }
+        }
+        catch (Exception e)
+        {
+            return false;
+        }
+    }
+
+    public boolean createPrincipal(Principal principal, char[] password)
+    {
+        if (_users.get(principal.getName()) != null)
+        {
+            return false;
+        }
+
+        PlainUser user = new PlainUser(principal.getName(), password);
+
+        try
+        {
+            _userUpdate.lock();
+            _users.put(user.getName(), user);
+
+            try
+            {
+                savePasswordFile();
+                return true;
+            }
+            catch (IOException e)
+            {
+                //remove the use on failure.
+                _users.remove(user.getName());
+                _logger.warn("Unable to create user '" + user.getName());
+                return false;
+            }
+        }
+        finally
+        {
+            if (_userUpdate.isHeldByCurrentThread())
+            {
+                _userUpdate.unlock();
+            }
+        }
+    }
+
+    public boolean deletePrincipal(Principal principal) throws AccountNotFoundException
+    {
+        PlainUser user = _users.get(principal.getName());
+
+        if (user == null)
+        {
+            throw new AccountNotFoundException(principal.getName());
+        }
+
+        try
+        {
+            _userUpdate.lock();
+            user.delete();
+
+            try
+            {
+                savePasswordFile();
+            }
+            catch (IOException e)
+            {
+                _logger.error("Unable to remove user '" + user.getName() + "' from password file.");
+                return false;
+            }
+
+            _users.remove(user.getName());
+        }
+        finally
+        {
+            if (_userUpdate.isHeldByCurrentThread())
+            {
+                _userUpdate.unlock();
+            }
+        }
+
+        return true;
+    }
+
+    public Map<String, AuthenticationProviderInitialiser> getMechanisms()
+    {
+        return _saslServers;
+    }
+
+    public List<Principal> getUsers()
+    {
+        return new LinkedList<Principal>(_users.values());
+    }
+
+    public Principal getUser(String username)
+    {
+        if (_users.containsKey(username))
+        {
+            return new UsernamePrincipal(username);
+        }
+        return null;
+    }
+
+    private boolean compareCharArray(char[] a, char[] b)
+    {
+        boolean equal = false;
+        if (a.length == b.length)
+        {
+            equal = true;
+            int index = 0;
+            while (equal && index < a.length)
+            {
+                equal = a[index] == b[index];
+                index++;
+            }
+        }
+        return equal;
+    }
+
+
+    /**
+     * Looks up the password for a specified user in the password file. Note this code is <b>not</b> secure since it
+     * creates strings of passwords. It should be modified to create only char arrays which get nulled out.
+     *
+     * @param name The principal name to lookup
+     *
+     * @return a char[] for use in SASL.
+     */
+    private char[] lookupPassword(String name)
+    {
+        PlainUser user = _users.get(name);
+        if (user == null)
+        {
+            return null;
+        }
+        else
+        {
+            return user.getPassword();
+        }
+    }
+
+    private void loadPasswordFile() throws IOException
+    {
+        try
+        {
+            _userUpdate.lock();
+            _users.clear();
+
+            BufferedReader reader = null;
+            try
+            {
+                reader = new BufferedReader(new FileReader(_passwordFile));
+                String line;
+
+                while ((line = reader.readLine()) != null)
+                {
+                    String[] result = _regexp.split(line);
+                    if (result == null || result.length < 2 || result[0].startsWith("#"))
+                    {
+                        continue;
+                    }
+
+                    PlainUser user = new PlainUser(result);
+                    _logger.info("Created user:" + user);
+                    _users.put(user.getName(), user);
+                }
+            }
+            finally
+            {
+                if (reader != null)
+                {
+                    reader.close();
+                }
+            }
+        }
+        finally
+        {
+            if (_userUpdate.isHeldByCurrentThread())
+            {
+                _userUpdate.unlock();
+            }
+        }
+    }
+
+    private void savePasswordFile() throws IOException
+    {
+        try
+        {
+            _userUpdate.lock();
+
+            BufferedReader reader = null;
+            PrintStream writer = null;
+            File tmp = File.createTempFile(_passwordFile.getName(), ".tmp");
+
+            try
+            {
+                writer = new PrintStream(tmp);
+                reader = new BufferedReader(new FileReader(_passwordFile));
+                String line;
+
+                while ((line = reader.readLine()) != null)
+                {
+                    String[] result = _regexp.split(line);
+                    if (result == null || result.length < 2 || result[0].startsWith("#"))
+                    {
+                        writer.write(line.getBytes(DEFAULT_ENCODING));
+                        writer.println();
+                        continue;
+                    }
+
+                    PlainUser user = _users.get(result[0]);
+
+                    if (user == null)
+                    {
+                        writer.write(line.getBytes(DEFAULT_ENCODING));
+                        writer.println();
+                    }
+                    else if (!user.isDeleted())
+                    {
+                        if (!user.isModified())
+                        {
+                            writer.write(line.getBytes(DEFAULT_ENCODING));
+                            writer.println();
+                        }
+                        else
+                        {
+                            byte[] password = user.getPasswordBytes();
+
+                            writer.write((user.getName() + ":").getBytes(DEFAULT_ENCODING));
+                            writer.write(password);
+                            writer.println();
+
+                            user.saved();
+                        }
+                    }
+                }
+
+                for (PlainUser user : _users.values())
+                {
+                    if (user.isModified())
+                    {
+                        byte[] password;
+                        password = user.getPasswordBytes();
+                        writer.write((user.getName() + ":").getBytes(DEFAULT_ENCODING));
+                        writer.write(password);
+                        writer.println();
+                        user.saved();
+                    }
+                }
+            }
+            finally
+            {
+                if (reader != null)
+                {
+                    reader.close();
+                }
+
+                if (writer != null)
+                {
+                    writer.close();
+                }
+
+                // Swap temp file to main password file.
+                File old = new File(_passwordFile.getAbsoluteFile() + ".old");
+                if (old.exists())
+                {
+                    old.delete();
+                }
+                _passwordFile.renameTo(old);
+                tmp.renameTo(_passwordFile);
+                tmp.delete();
+            }
+        }
+        finally
+        {
+            if (_userUpdate.isHeldByCurrentThread())
+            {
+                _userUpdate.unlock();
+            }
+        }
+    }
+    
+    public void reload() throws IOException
+    {
+        loadPasswordFile();
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PlainUser.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PlainUser.java
new file mode 100644
index 0000000000..46a78a55aa
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PlainUser.java
@@ -0,0 +1,106 @@
+/*
+*
+ * 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.database;
+
+import org.apache.log4j.Logger;
+
+import java.security.Principal;
+
+public class PlainUser implements Principal
+{
+    private String _name;
+    private char[] _password;
+    private boolean _modified = false;
+    private boolean _deleted = false;
+
+    PlainUser(String[] data)
+    {
+        if (data.length != 2)
+        {
+            throw new IllegalArgumentException("User Data should be length 2, username, password");
+        }
+
+        _name = data[0];
+
+        _password = data[1].toCharArray();
+
+    }
+
+    public PlainUser(String name, char[] password)
+    {
+        _name = name;
+        _password = password;
+        _modified = true;
+    }
+
+    public String getName()
+    {
+        return _name;
+    }
+
+    public String toString()
+    {
+        return _name;
+    }
+
+    char[] getPassword()
+    {
+        return _password;
+    }
+    
+    byte[] getPasswordBytes()
+    {
+        byte[] byteArray = new byte[_password.length];
+        int index = 0;
+        for (char c : _password)
+        {
+            byteArray[index++] = (byte) c;
+        }
+        return byteArray;
+    }
+
+    void setPassword(char[] password)
+    {
+        _password = password;
+        _modified = true;
+    }
+
+    public boolean isModified()
+    {
+        return _modified;
+    }
+
+    public boolean isDeleted()
+    {
+        return _deleted;
+    }
+
+    public void delete()
+    {
+        _deleted = true;
+    }
+
+    public void saved()
+    {
+        _modified = false;
+    }
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PrincipalDatabase.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PrincipalDatabase.java
new file mode 100644
index 0000000000..ef37e043a6
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PrincipalDatabase.java
@@ -0,0 +1,105 @@
+/*
+ *
+ * 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.database;
+
+import org.apache.qpid.server.security.auth.sasl.AuthenticationProviderInitialiser;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.security.Principal;
+import java.util.Map;
+import java.util.List;
+
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.login.AccountNotFoundException;
+
+/** Represents a "user database" which is really a way of storing principals (i.e. usernames) and passwords. */
+public interface PrincipalDatabase
+{
+    /**
+     * Set the password for a given principal in the specified callback. This is used for certain SASL providers. The
+     * user database implementation should look up the password in any way it chooses and set it in the callback by
+     * calling its setPassword method.
+     *
+     * @param principal the principal
+     * @param callback  the password callback that wants to receive the password
+     *
+     * @throws AccountNotFoundException if the account for specified principal could not be found
+     * @throws IOException              if there was an error looking up the principal
+     */
+    void setPassword(Principal principal, PasswordCallback callback)
+            throws IOException, AccountNotFoundException;
+
+     /**
+     * Used to verify that the presented Password is correct. Currently only used by Management Console
+     * @param principal The principal to authenticate
+     * @param password The password to check
+     * @return true if password is correct
+     * @throws AccountNotFoundException if the principal cannot be found
+     */
+    boolean verifyPassword(String principal, char[] password)
+            throws AccountNotFoundException;
+
+    /**
+     * Update(Change) the password for the given principal
+     * @param principal Who's password is to be changed
+     * @param password The new password to use
+     * @return True if change was successful
+     * @throws AccountNotFoundException If the given principal doesn't exist in the Database
+     */
+    boolean updatePassword(Principal principal, char[] password)
+            throws AccountNotFoundException;
+
+    /**
+     * Create a new principal in the database
+     * @param principal The principal to create
+     * @param password The password to set for the principal
+     * @return True on a successful creation
+     */
+    boolean createPrincipal(Principal principal, char[] password);
+
+    /**
+     * Delete a principal
+     * @param principal The principal to delete
+     * @return True on a successful creation
+     * @throws AccountNotFoundException If the given principal doesn't exist in the Database
+     */
+    boolean deletePrincipal(Principal principal)
+            throws AccountNotFoundException;
+
+    /**
+     * Get the principal from the database with the given username
+     * @param username of the principal to lookup
+     * @return The Principal object for the given username or null if not found.
+     */
+    Principal getUser(String username);
+
+    /**
+     * Reload the database to its ensure contents are up to date
+     * @throws IOException If there was an error reloading the database
+     */
+    void reload() throws IOException;
+
+    public Map<String, AuthenticationProviderInitialiser> getMechanisms();
+
+
+    List<Principal> getUsers();
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PrincipalDatabaseManager.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PrincipalDatabaseManager.java
new file mode 100644
index 0000000000..f9882f8810
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PrincipalDatabaseManager.java
@@ -0,0 +1,35 @@
+/*
+ *  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.database;
+
+import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.qpid.server.security.auth.database.PrincipalDatabase;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+
+import java.util.Map;
+
+public interface PrincipalDatabaseManager
+{
+    public Map<String, PrincipalDatabase> getDatabases();
+
+    public void initialiseManagement(ServerConfiguration _configuration) throws ConfigurationException;
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PropertiesPrincipalDatabase.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PropertiesPrincipalDatabase.java
new file mode 100644
index 0000000000..ff8851306f
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PropertiesPrincipalDatabase.java
@@ -0,0 +1,169 @@
+/*
+ *  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.database;
+
+import org.apache.qpid.server.security.auth.sasl.AuthenticationProviderInitialiser;
+import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
+import org.apache.qpid.server.security.auth.sasl.crammd5.CRAMMD5Initialiser;
+import org.apache.qpid.server.security.auth.sasl.plain.PlainInitialiser;
+
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.login.AccountNotFoundException;
+import java.util.Properties;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.List;
+import java.util.LinkedList;
+import java.security.Principal;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+
+public class PropertiesPrincipalDatabase implements PrincipalDatabase
+{
+    private Properties _users;
+
+    private Map<String, AuthenticationProviderInitialiser> _saslServers;
+
+    public PropertiesPrincipalDatabase(Properties users)
+    {
+        _users = users;
+
+        _saslServers = new HashMap<String, AuthenticationProviderInitialiser>();
+
+        /**
+         *  Create Authenticators for Properties Principal Database.
+         */
+
+        //  Accept MD5 incomming and use plain comparison with the file
+        PlainInitialiser cram = new PlainInitialiser();
+        cram.initialise(this);
+        // Accept Plain incomming and hash it for comparison to the file.
+        CRAMMD5Initialiser plain = new CRAMMD5Initialiser();
+        plain.initialise(this, CRAMMD5Initialiser.HashDirection.INCOMMING);
+
+        _saslServers.put(plain.getMechanismName(), cram);
+        _saslServers.put(cram.getMechanismName(), plain);
+    }
+
+    public void setPassword(Principal principal, PasswordCallback callback) throws IOException, AccountNotFoundException
+    {
+        if (principal == null)
+        {
+            throw new IllegalArgumentException("principal must not be null");
+        }
+
+
+
+        final String pwd = _users.getProperty(principal.getName());
+
+        if (pwd != null)
+        {
+            callback.setPassword(pwd.toCharArray());
+        }
+        else
+        {
+            throw new AccountNotFoundException("No account found for principal " + principal);
+        }
+    }
+
+    public boolean verifyPassword(String principal, char[] password) throws AccountNotFoundException
+    {
+        //fixme this is not correct as toCharArray is not safe based on the type of string.
+        char[] pwd = _users.getProperty(principal).toCharArray();
+
+        return compareCharArray(pwd, password);
+    }
+
+    public boolean updatePassword(Principal principal, char[] password) throws AccountNotFoundException
+    {
+        return false; // updates denied
+    }
+
+    public boolean createPrincipal(Principal principal, char[] password)
+    {
+        return false; // updates denied
+    }
+
+    public boolean deletePrincipal(Principal principal) throws AccountNotFoundException
+    {
+        return false; // updates denied
+    }
+
+    private boolean compareCharArray(char[] a, char[] b)
+    {
+        boolean equal = false;
+        if (a.length == b.length)
+        {
+            equal = true;
+            int index = 0;
+            while (equal && index < a.length)
+            {
+                equal = a[index] == b[index];
+                index++;
+            }
+        }
+        return equal;
+    }
+
+    private char[] convertPassword(String password) throws UnsupportedEncodingException
+    {
+        byte[] passwdBytes = password.getBytes("utf-8");
+
+        char[] passwd = new char[passwdBytes.length];
+
+        int index = 0;
+
+        for (byte b : passwdBytes)
+        {
+            passwd[index++] = (char) b;
+        }
+
+        return passwd;
+    }
+
+
+    public Map<String, AuthenticationProviderInitialiser> getMechanisms()
+    {
+        return _saslServers;
+    }
+
+    public List<Principal> getUsers()
+    {
+        return new LinkedList<Principal>(); //todo
+    }
+
+    public Principal getUser(String username)
+    {
+        if (_users.getProperty(username) != null)
+        {
+            return new UsernamePrincipal(username);
+        }
+        else
+        {
+            return null;
+        }
+    }
+
+    public void reload() throws IOException
+    {
+        //No file to update from, so do nothing.
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PropertiesPrincipalDatabaseManager.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PropertiesPrincipalDatabaseManager.java
new file mode 100644
index 0000000000..4efe381a8b
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/database/PropertiesPrincipalDatabaseManager.java
@@ -0,0 +1,50 @@
+/*
+ *  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.database;
+
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.qpid.server.configuration.ServerConfiguration;
+
+import java.util.Map;
+import java.util.Properties;
+import java.util.HashMap;
+
+public class PropertiesPrincipalDatabaseManager implements PrincipalDatabaseManager
+{
+
+    Map<String, PrincipalDatabase> _databases = new HashMap<String, PrincipalDatabase>();
+
+    public PropertiesPrincipalDatabaseManager(String name, Properties users)
+    {
+        _databases.put(name, new PropertiesPrincipalDatabase(users));
+    }
+
+    public Map<String, PrincipalDatabase> getDatabases()
+    {
+        return _databases;
+    }
+
+    @Override
+    public void initialiseManagement(ServerConfiguration _configuration) throws ConfigurationException
+    {
+        //todo
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/AuthenticationManager.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/AuthenticationManager.java
new file mode 100644
index 0000000000..d1803124a7
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/AuthenticationManager.java
@@ -0,0 +1,38 @@
+/*
+ *
+ * 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.manager;
+
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.security.auth.AuthenticationResult;
+
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+
+public interface AuthenticationManager
+{
+    String getMechanisms();
+
+    SaslServer createSaslServer(String mechanism, String localFQDN) throws SaslException;
+
+    AuthenticationResult authenticate(SaslServer server, byte[] response);
+
+    void close();
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManager.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManager.java
new file mode 100644
index 0000000000..98c060599a
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManager.java
@@ -0,0 +1,237 @@
+/*
+ *  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.manager;
+
+import org.apache.log4j.Logger;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.qpid.server.configuration.VirtualHostConfiguration;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.security.auth.manager.AuthenticationManager;
+import org.apache.qpid.server.security.auth.database.PrincipalDatabase;
+import org.apache.qpid.server.security.auth.sasl.JCAProvider;
+import org.apache.qpid.server.security.auth.sasl.AuthenticationProviderInitialiser;
+import org.apache.qpid.server.security.auth.AuthenticationResult;
+
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.SaslServerFactory;
+import javax.security.sasl.SaslServer;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.Sasl;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.TreeMap;
+import java.security.Security;
+
+public class PrincipalDatabaseAuthenticationManager implements AuthenticationManager
+{
+    private static final Logger _logger = Logger.getLogger(PrincipalDatabaseAuthenticationManager.class);
+
+    /** The list of mechanisms, in the order in which they are configured (i.e. preferred order) */
+    private String _mechanisms;
+
+    /** Maps from the mechanism to the callback handler to use for handling those requests */
+    private Map<String, CallbackHandler> _callbackHandlerMap = new HashMap<String, CallbackHandler>();
+
+    /**
+     * Maps from the mechanism to the properties used to initialise the server. See the method Sasl.createSaslServer for
+     * details of the use of these properties. This map is populated during initialisation of each provider.
+     */
+    private Map<String, Map<String, ?>> _serverCreationProperties = new HashMap<String, Map<String, ?>>();
+
+    private AuthenticationManager _default = null;
+    /** The name for the required SASL Server mechanisms */
+    public static final String PROVIDER_NAME= "AMQSASLProvider-Server";
+
+    public PrincipalDatabaseAuthenticationManager(String name, VirtualHostConfiguration hostConfig) throws Exception
+    {
+        _logger.info("Initialising " + (name == null ? "Default" : "'" + name + "'")
+                     + " PrincipleDatabase authentication manager.");
+
+        // Fixme This should be done per Vhost but allowing global hack isn't right but ...
+        // required as authentication is done before Vhost selection
+
+        Map<String, Class<? extends SaslServerFactory>> providerMap = new TreeMap<String, Class<? extends SaslServerFactory>>();
+
+
+        if (name == null || hostConfig == null)
+        {
+            initialiseAuthenticationMechanisms(providerMap, ApplicationRegistry.getInstance().getDatabaseManager().getDatabases());
+        }
+        else
+        {
+            String databaseName = hostConfig.getAuthenticationDatabase();
+
+            if (databaseName == null)
+            {
+
+                _default = ApplicationRegistry.getInstance().getAuthenticationManager();
+                return;
+            }
+            else
+            {
+                PrincipalDatabase database = ApplicationRegistry.getInstance().getDatabaseManager().getDatabases().get(databaseName);
+
+                if (database == null)
+                {
+                    throw new ConfigurationException("Requested database:" + databaseName + " was not found");
+                }
+
+                initialiseAuthenticationMechanisms(providerMap, database);
+            }
+        }
+
+        if (providerMap.size() > 0)
+        {
+            // Ensure we are used before the defaults
+            if (Security.insertProviderAt(new JCAProvider(PROVIDER_NAME, providerMap), 1) == -1)
+            {
+                _logger.error("Unable to load custom SASL providers. Qpid custom SASL authenticators unavailable.");
+            }
+            else
+            {
+                _logger.info("Additional SASL providers successfully registered.");
+            }
+
+        }
+        else
+        {
+            _logger.warn("No additional SASL providers registered.");
+        }
+
+    }
+
+
+    private void initialiseAuthenticationMechanisms(Map<String, Class<? extends SaslServerFactory>> providerMap, Map<String, PrincipalDatabase> databases) throws Exception
+    {
+        if (databases.size() > 1)
+        {
+            _logger.warn("More than one principle database provided currently authentication mechanism will override each other.");
+        }
+
+        for (Map.Entry<String, PrincipalDatabase> entry : databases.entrySet())
+        {
+            // fixme As the database now provide the mechanisms they support, they will ...
+            // overwrite each other in the map. There should only be one database per vhost.
+            // But currently we must have authentication before vhost definition.
+            initialiseAuthenticationMechanisms(providerMap, entry.getValue());
+        }
+    }
+
+    private void initialiseAuthenticationMechanisms(Map<String, Class<? extends SaslServerFactory>> providerMap, PrincipalDatabase database) throws Exception
+    {
+        if (database == null || database.getMechanisms().size() == 0)
+        {
+            _logger.warn("No Database or no mechanisms to initialise authentication");
+            return;
+        }
+
+        for (Map.Entry<String, AuthenticationProviderInitialiser> mechanism : database.getMechanisms().entrySet())
+        {
+            initialiseAuthenticationMechanism(mechanism.getKey(), mechanism.getValue(), providerMap);
+        }
+    }
+
+    private void initialiseAuthenticationMechanism(String mechanism, AuthenticationProviderInitialiser initialiser,
+                                                   Map<String, Class<? extends SaslServerFactory>> providerMap)
+            throws Exception
+    {
+        if (_mechanisms == null)
+        {
+            _mechanisms = mechanism;
+        }
+        else
+        {
+            // simple append should be fine since the number of mechanisms is small and this is a one time initialisation
+            _mechanisms = _mechanisms + " " + mechanism;
+        }
+        _callbackHandlerMap.put(mechanism, initialiser.getCallbackHandler());
+        _serverCreationProperties.put(mechanism, initialiser.getProperties());
+        Class<? extends SaslServerFactory> factory = initialiser.getServerFactoryClassForJCARegistration();
+        if (factory != null)
+        {
+            providerMap.put(mechanism, factory);
+        }
+        _logger.info("Initialised " + mechanism + " SASL provider successfully");
+    }
+
+    public String getMechanisms()
+    {
+        if (_default != null)
+        {
+            // Use the default AuthenticationManager if present
+            return _default.getMechanisms();
+        }
+        else
+        {
+            return _mechanisms;
+        }
+    }
+
+    public SaslServer createSaslServer(String mechanism, String localFQDN) throws SaslException
+    {
+        if (_default != null)
+        {
+            // Use the default AuthenticationManager if present
+            return _default.createSaslServer(mechanism, localFQDN);
+        }
+        else
+        {
+            return Sasl.createSaslServer(mechanism, "AMQP", localFQDN, _serverCreationProperties.get(mechanism),
+                                         _callbackHandlerMap.get(mechanism));
+        }
+
+    }
+
+    public AuthenticationResult authenticate(SaslServer server, byte[] response)
+    {
+        // Use the default AuthenticationManager if present
+        if (_default != null)
+        {
+            return _default.authenticate(server, response);
+        }
+
+
+        try
+        {
+            // Process response from the client
+            byte[] challenge = server.evaluateResponse(response != null ? response : new byte[0]);
+
+            if (server.isComplete())
+            {
+                return new AuthenticationResult(challenge, AuthenticationResult.AuthenticationStatus.SUCCESS);
+            }
+            else
+            {
+                return new AuthenticationResult(challenge, AuthenticationResult.AuthenticationStatus.CONTINUE);
+            }
+        }
+        catch (SaslException e)
+        {
+            return new AuthenticationResult(AuthenticationResult.AuthenticationStatus.ERROR, e);
+        }
+    }
+
+    public void close()
+    {
+        Security.removeProvider(PROVIDER_NAME);
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/rmi/RMIPasswordAuthenticator.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/rmi/RMIPasswordAuthenticator.java
new file mode 100644
index 0000000000..77040e896c
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/rmi/RMIPasswordAuthenticator.java
@@ -0,0 +1,119 @@
+/*
+ *
+ * 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.rmi;
+
+import java.util.Collections;
+
+import javax.management.remote.JMXAuthenticator;
+import javax.management.remote.JMXPrincipal;
+import javax.security.auth.Subject;
+import javax.security.auth.login.AccountNotFoundException;
+
+import org.apache.qpid.server.security.auth.database.PrincipalDatabase;
+
+public class RMIPasswordAuthenticator implements JMXAuthenticator
+{
+    static final String UNABLE_TO_LOOKUP = "The broker was unable to lookup the user details";
+    static final String SHOULD_BE_STRING_ARRAY = "User details should be String[]";
+    static final String SHOULD_HAVE_2_ELEMENTS = "User details should have 2 elements, username, password";
+    static final String SHOULD_BE_NON_NULL = "Supplied username and password should be non-null";
+    static final String INVALID_CREDENTIALS = "Invalid user details supplied";
+    static final String CREDENTIALS_REQUIRED = "User details are required. " +
+    		            "Please ensure you are using an up to date management console to connect.";
+    
+    private PrincipalDatabase _db = null;
+
+    public RMIPasswordAuthenticator()
+    {
+    }
+    
+    public void setPrincipalDatabase(PrincipalDatabase pd)
+    {
+        this._db = pd;
+    }
+
+    public Subject authenticate(Object credentials) throws SecurityException
+    {
+        // Verify that credential's are of type String[].
+        if (!(credentials instanceof String[]))
+        {
+            if (credentials == null)
+            {
+                throw new SecurityException(CREDENTIALS_REQUIRED);
+            }
+            else
+            {
+                throw new SecurityException(SHOULD_BE_STRING_ARRAY);
+            }
+        }
+
+        // Verify that required number of credential's.
+        final String[] userCredentials = (String[]) credentials;
+        if (userCredentials.length != 2)
+        {
+            throw new SecurityException(SHOULD_HAVE_2_ELEMENTS);
+        }
+
+        String username = (String) userCredentials[0];
+        String password = (String) userCredentials[1];
+
+        // Verify that all required credential's are actually present.
+        if (username == null || password == null)
+        {
+            throw new SecurityException(SHOULD_BE_NON_NULL);
+        }
+        
+        // Verify that a PD has been set.
+        if (_db == null)
+        {
+            throw new SecurityException(UNABLE_TO_LOOKUP);
+        }
+        
+        boolean authenticated = false;
+
+        // Perform authentication
+        try
+        {
+            if (_db.verifyPassword(username, password.toCharArray()))
+            {            
+                authenticated = true;
+            }
+        }
+        catch (AccountNotFoundException e)
+        {
+            throw new SecurityException(INVALID_CREDENTIALS);
+        }
+
+        if (authenticated)
+        {
+            //credential's check out, return the appropriate JAAS Subject
+            return new Subject(true,
+                    Collections.singleton(new JMXPrincipal(username)),
+                    Collections.EMPTY_SET,
+                    Collections.EMPTY_SET);
+        }
+        else
+        {
+            throw new SecurityException(INVALID_CREDENTIALS);
+        }
+    }
+
+}
\ No newline at end of file
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/AuthenticationProviderInitialiser.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/AuthenticationProviderInitialiser.java
new file mode 100644
index 0000000000..89e545d6f5
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/AuthenticationProviderInitialiser.java
@@ -0,0 +1,76 @@
+/*
+ *
+ * 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.util.Map;
+
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.SaslServerFactory;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.qpid.server.security.auth.database.PrincipalDatabase;
+
+public interface AuthenticationProviderInitialiser
+{
+    /**
+     * @return the mechanism's name. This will be used in the list of mechanism's advertised to the
+     * client.
+     */
+    String getMechanismName();
+
+    /**
+     * Initialise the authentication provider.
+     * @param baseConfigPath the path in the config file that points to any config options for this provider. Each
+     * provider can have its own set of configuration options
+     * @param configuration the Apache Commons Configuration instance used to configure this provider
+     * @param principalDatabases the set of principal databases that are available
+     * @throws Exception needs refined Exception is too broad.
+     */
+    void initialise(String baseConfigPath, Configuration configuration,
+                    Map<String, PrincipalDatabase> principalDatabases) throws Exception;
+
+    /**
+     * Initialise the authentication provider.     
+     * @param db The principal database to initialise with
+     */
+    void initialise(PrincipalDatabase db);
+
+
+    /**
+     * @return the callback handler that should be used to process authentication requests for this mechanism. This will
+     * be called after initialise and will be stored by the authentication manager. The callback handler <b>must</b> be
+     * fully threadsafe.
+     */
+    CallbackHandler getCallbackHandler();
+
+    /**
+     * Get the properties that must be passed in to the Sasl.createSaslServer method.
+     * @return the properties, which may be null
+     */
+    Map<String, ?> getProperties();
+
+    /**
+     * Get the class that is the server factory. This is used for the JCA registration.
+     * @return null if no JCA registration is required, otherwise return the class
+     * that will be used in JCA registration
+     */
+    Class<? extends SaslServerFactory> getServerFactoryClassForJCARegistration();
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/JCAProvider.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/JCAProvider.java
new file mode 100644
index 0000000000..d6a09d8217
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/JCAProvider.java
@@ -0,0 +1,46 @@
+/*
+ *
+ * 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.Provider;
+import java.security.Security;
+import java.util.Map;
+
+import javax.security.sasl.SaslServerFactory;
+
+public final class JCAProvider extends Provider
+{
+    public JCAProvider(String name, Map<String, Class<? extends SaslServerFactory>> providerMap)
+    {
+        super(name, 1.0, "A JCA provider that registers all " +
+              "AMQ SASL providers that want to be registered");
+        register(providerMap);
+    }
+
+    private void register(Map<String, Class<? extends SaslServerFactory>> providerMap)
+    {
+        for (Map.Entry<String, Class<? extends SaslServerFactory>> me :
+             providerMap.entrySet())
+        {
+            put("SaslServerFactory." + me.getKey(), me.getValue().getName());
+        }
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/UsernamePasswordInitialiser.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/UsernamePasswordInitialiser.java
new file mode 100644
index 0000000000..dd0bd096c3
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/UsernamePasswordInitialiser.java
@@ -0,0 +1,123 @@
+/*
+ *
+ * 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.io.IOException;
+import java.security.Principal;
+import java.util.Map;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.auth.login.AccountNotFoundException;
+import javax.security.sasl.AuthorizeCallback;
+
+import org.apache.commons.configuration.Configuration;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.server.security.auth.database.PrincipalDatabase;
+import org.apache.qpid.server.security.auth.sasl.AuthenticationProviderInitialiser;
+import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
+
+public abstract class UsernamePasswordInitialiser implements AuthenticationProviderInitialiser
+{
+    protected static final Logger _logger = Logger.getLogger(UsernamePasswordInitialiser.class);
+
+    private ServerCallbackHandler _callbackHandler;
+
+    private class ServerCallbackHandler implements CallbackHandler
+    {
+        private final PrincipalDatabase _principalDatabase;
+
+        protected ServerCallbackHandler(PrincipalDatabase database)
+        {
+            _principalDatabase = database;
+        }
+
+        public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException
+        {
+            Principal username = null;
+            for (Callback callback : callbacks)
+            {
+                if (callback instanceof NameCallback)
+                {
+                    username = new UsernamePrincipal(((NameCallback) callback).getDefaultName());
+                }
+                else if (callback instanceof PasswordCallback)
+                {
+                    try
+                    {
+                        _principalDatabase.setPassword(username, (PasswordCallback) callback);
+                    }
+                    catch (AccountNotFoundException e)
+                    {
+                        // very annoyingly the callback handler does not throw anything more appropriate than
+                        // IOException
+                        IOException ioe = new IOException("Error looking up user " + e);
+                        ioe.initCause(e);
+                        throw ioe;
+                    }
+                }
+                else if (callback instanceof AuthorizeCallback)
+                {
+                    ((AuthorizeCallback) callback).setAuthorized(true);
+                }
+                else
+                {
+                    throw new UnsupportedCallbackException(callback);
+                }
+            }
+        }
+    }
+
+    public void initialise(String baseConfigPath, Configuration configuration,
+        Map<String, PrincipalDatabase> principalDatabases) throws Exception
+    {
+        String principalDatabaseName = configuration.getString(baseConfigPath + ".principal-database");
+        PrincipalDatabase db = principalDatabases.get(principalDatabaseName);
+
+        initialise(db);
+    }
+
+    public void initialise(PrincipalDatabase db)
+    {
+        if (db == null)
+        {
+            throw new NullPointerException("Cannot initialise with a null Principal database.");
+        }
+
+        _callbackHandler = new ServerCallbackHandler(db);
+    }
+
+    public CallbackHandler getCallbackHandler()
+    {
+        return _callbackHandler;
+    }
+
+    public Map<String, ?> getProperties()
+    {
+        // there are no properties required for the CRAM-MD5 implementation
+        return null;
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/UsernamePrincipal.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/UsernamePrincipal.java
new file mode 100644
index 0000000000..d7c8383690
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/UsernamePrincipal.java
@@ -0,0 +1,44 @@
+/*
+ *
+ * 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;
+
+/** A principal that is just a wrapper for a simple username. */
+public class UsernamePrincipal implements Principal
+{
+    private String _name;
+
+    public UsernamePrincipal(String name)
+    {
+        _name = name;
+    }
+
+    public String getName()
+    {
+        return _name;
+    }
+
+    public String toString()
+    {
+        return _name;
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainInitialiser.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainInitialiser.java
new file mode 100644
index 0000000000..7acc6322d1
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainInitialiser.java
@@ -0,0 +1,38 @@
+/*
+ *
+ * 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.amqplain;
+
+import javax.security.sasl.SaslServerFactory;
+
+import org.apache.qpid.server.security.auth.sasl.UsernamePasswordInitialiser;
+
+public class AmqPlainInitialiser extends UsernamePasswordInitialiser
+{
+    public String getMechanismName()
+    {
+        return "AMQPLAIN";
+    }
+
+    public Class<? extends SaslServerFactory> getServerFactoryClassForJCARegistration()
+    {
+        return AmqPlainSaslServerFactory.class;
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainSaslServer.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainSaslServer.java
new file mode 100644
index 0000000000..9f56b8521a
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainSaslServer.java
@@ -0,0 +1,132 @@
+/*
+ *
+ * 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.amqplain;
+
+import java.io.IOException;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.sasl.AuthorizeCallback;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+
+import org.apache.mina.common.ByteBuffer;
+import org.apache.qpid.framing.AMQFrameDecodingException;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.framing.FieldTableFactory;
+
+public class AmqPlainSaslServer implements SaslServer
+{
+    public static final String MECHANISM = "AMQPLAIN";
+
+    private CallbackHandler _cbh;
+
+    private String _authorizationId;
+
+    private boolean _complete = false;
+
+    public AmqPlainSaslServer(CallbackHandler cbh)
+    {
+        _cbh = cbh;
+    }
+
+    public String getMechanismName()
+    {
+        return MECHANISM;
+    }
+
+    public byte[] evaluateResponse(byte[] response) throws SaslException
+    {
+        try
+        {
+            final FieldTable ft = FieldTableFactory.newFieldTable(ByteBuffer.wrap(response), response.length);
+            String username = (String) ft.getString("LOGIN");
+            // we do not care about the prompt but it throws if null
+            NameCallback nameCb = new NameCallback("prompt", username);
+            // we do not care about the prompt but it throws if null
+            PasswordCallback passwordCb = new PasswordCallback("prompt", false);
+            // TODO: should not get pwd as a String but as a char array...
+            String pwd = (String) ft.getString("PASSWORD");
+            AuthorizeCallback authzCb = new AuthorizeCallback(username, username);
+            Callback[] callbacks = new Callback[]{nameCb, passwordCb, authzCb};
+            _cbh.handle(callbacks);
+            String storedPwd = new String(passwordCb.getPassword());
+            if (storedPwd.equals(pwd))
+            {
+                _complete = true;
+            }
+            if (authzCb.isAuthorized() && _complete)
+            {
+                _authorizationId = authzCb.getAuthenticationID();
+                return null;
+            }
+            else
+            {
+                throw new SaslException("Authentication failed");
+            }
+        }
+        catch (AMQFrameDecodingException e)
+        {
+            throw new SaslException("Unable to decode response: " + e, e);
+        }
+        catch (IOException e)
+        {
+            throw new SaslException("Error processing data: " + e, e);
+        }
+        catch (UnsupportedCallbackException e)
+        {
+            throw new SaslException("Unable to obtain data from callback handler: " + e, e);
+        }
+    }
+
+    public boolean isComplete()
+    {
+        return _complete;
+    }
+
+    public String getAuthorizationID()
+    {
+        return _authorizationId;
+    }
+
+    public byte[] unwrap(byte[] incoming, int offset, int len) throws SaslException
+    {
+        throw new SaslException("Unsupported operation");
+    }
+
+    public byte[] wrap(byte[] outgoing, int offset, int len) throws SaslException
+    {
+        throw new SaslException("Unsupported operation");
+    }
+
+    public Object getNegotiatedProperty(String propName)
+    {
+        return null;
+    }
+
+    public void dispose() throws SaslException
+    {
+        _cbh = null;
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainSaslServerFactory.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainSaslServerFactory.java
new file mode 100644
index 0000000000..67d20136bf
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/amqplain/AmqPlainSaslServerFactory.java
@@ -0,0 +1,60 @@
+/*
+ *
+ * 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.amqplain;
+
+import java.util.Map;
+
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import javax.security.sasl.SaslServerFactory;
+
+public class AmqPlainSaslServerFactory implements SaslServerFactory
+{
+    public SaslServer createSaslServer(String mechanism, String protocol, String serverName, Map props,
+                                       CallbackHandler cbh) throws SaslException
+    {
+        if (AmqPlainSaslServer.MECHANISM.equals(mechanism))
+        {
+            return new AmqPlainSaslServer(cbh);
+        }
+        else
+        {
+            return null;
+        }
+    }
+
+    public String[] getMechanismNames(Map props)
+    {
+        if (props.containsKey(Sasl.POLICY_NOPLAINTEXT) ||
+            props.containsKey(Sasl.POLICY_NODICTIONARY) ||
+            props.containsKey(Sasl.POLICY_NOACTIVE))
+        {
+            // returned array must be non null according to interface documentation
+            return new String[0];
+        }
+        else
+        {
+            return new String[]{AmqPlainSaslServer.MECHANISM};
+        }
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HashedInitialiser.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HashedInitialiser.java
new file mode 100644
index 0000000000..97f9a4e91a
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HashedInitialiser.java
@@ -0,0 +1,50 @@
+/*
+ *
+ * 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.crammd5;
+
+import org.apache.qpid.server.security.auth.sasl.UsernamePasswordInitialiser;
+import org.apache.qpid.server.security.auth.database.PrincipalDatabase;
+
+import javax.security.sasl.SaslServerFactory;
+import java.util.Map;
+
+public class CRAMMD5HashedInitialiser extends UsernamePasswordInitialiser
+{
+    public String getMechanismName()
+    {
+        return CRAMMD5HashedSaslServer.MECHANISM;
+    }
+
+    public Class<? extends SaslServerFactory> getServerFactoryClassForJCARegistration()
+    {
+        return CRAMMD5HashedServerFactory.class;
+    }
+
+    public void initialise(PrincipalDatabase passwordFile)
+    {
+        super.initialise(passwordFile);
+    }
+
+    public Map<String, ?> getProperties()
+    {
+        return null;
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HashedSaslServer.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HashedSaslServer.java
new file mode 100644
index 0000000000..f6cab084ea
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HashedSaslServer.java
@@ -0,0 +1,105 @@
+/*
+ *  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.crammd5;
+
+import javax.security.sasl.SaslServer;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslServerFactory;
+import javax.security.auth.callback.CallbackHandler;
+import java.util.Enumeration;
+import java.util.Map;
+
+public class CRAMMD5HashedSaslServer implements SaslServer
+{
+    public static final String MECHANISM = "CRAM-MD5-HASHED";
+
+    private SaslServer _realServer;
+
+    public CRAMMD5HashedSaslServer(String mechanism, String protocol, String serverName, Map<String, ?> props,
+                                   CallbackHandler cbh) throws SaslException
+    {
+        Enumeration factories = Sasl.getSaslServerFactories();
+
+        while (factories.hasMoreElements())
+        {
+            SaslServerFactory factory = (SaslServerFactory) factories.nextElement();
+
+            if (factory instanceof CRAMMD5HashedServerFactory)
+            {
+                continue;
+            }
+            
+            String[] mechs = factory.getMechanismNames(props);
+
+            for (String mech : mechs)
+            {
+                if (mech.equals("CRAM-MD5"))
+                {
+                    _realServer = factory.createSaslServer("CRAM-MD5", protocol, serverName, props, cbh);
+                    return;
+                }
+            }
+        }
+
+        throw new RuntimeException("No default SaslServer found for mechanism:" + "CRAM-MD5");
+    }
+
+    public String getMechanismName()
+    {
+        return MECHANISM;
+    }
+
+    public byte[] evaluateResponse(byte[] response) throws SaslException
+    {
+        return _realServer.evaluateResponse(response);
+    }
+
+    public boolean isComplete()
+    {
+        return _realServer.isComplete();
+    }
+
+    public String getAuthorizationID()
+    {
+        return _realServer.getAuthorizationID();
+    }
+
+    public byte[] unwrap(byte[] incoming, int offset, int len) throws SaslException
+    {
+        return _realServer.unwrap(incoming, offset, len);
+    }
+
+    public byte[] wrap(byte[] outgoing, int offset, int len) throws SaslException
+    {
+        return _realServer.wrap(outgoing, offset, len);
+    }
+
+    public Object getNegotiatedProperty(String propName)
+    {
+        return _realServer.getNegotiatedProperty(propName);
+    }
+
+    public void dispose() throws SaslException
+    {
+        _realServer.dispose();
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HashedServerFactory.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HashedServerFactory.java
new file mode 100644
index 0000000000..5298b5cc63
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5HashedServerFactory.java
@@ -0,0 +1,61 @@
+/*
+ *
+ * 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.crammd5;
+
+import java.util.Map;
+
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import javax.security.sasl.SaslServerFactory;
+
+public class CRAMMD5HashedServerFactory implements SaslServerFactory
+{
+    public SaslServer createSaslServer(String mechanism, String protocol, String serverName, Map<String, ?> props,
+                                       CallbackHandler cbh) throws SaslException
+    {
+        if (mechanism.equals(CRAMMD5HashedSaslServer.MECHANISM))
+        {
+            return new CRAMMD5HashedSaslServer(mechanism, protocol, serverName, props, cbh);
+        }
+        else
+        {
+            return null;
+        }
+    }
+
+    public String[] getMechanismNames(Map props)
+    {
+        if (props != null)
+        {
+            if (props.containsKey(Sasl.POLICY_NOPLAINTEXT) ||
+                props.containsKey(Sasl.POLICY_NODICTIONARY) ||
+                props.containsKey(Sasl.POLICY_NOACTIVE))
+            {
+                // returned array must be non null according to interface documentation
+                return new String[0];
+            }
+        }
+
+        return new String[]{CRAMMD5HashedSaslServer.MECHANISM};
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5Initialiser.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5Initialiser.java
new file mode 100644
index 0000000000..264832888d
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/crammd5/CRAMMD5Initialiser.java
@@ -0,0 +1,71 @@
+/*
+ *
+ * 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.crammd5;
+
+import org.apache.qpid.server.security.auth.sasl.UsernamePasswordInitialiser;
+import org.apache.qpid.server.security.auth.database.PrincipalDatabase;
+
+import javax.security.sasl.SaslServerFactory;
+
+public class CRAMMD5Initialiser extends UsernamePasswordInitialiser
+{
+    private HashDirection _hashDirection;
+
+    public enum HashDirection
+    {
+        INCOMMING, PASSWORD_FILE
+    }
+
+
+    public String getMechanismName()
+    {
+        return "CRAM-MD5";
+    }
+
+    public Class<? extends SaslServerFactory> getServerFactoryClassForJCARegistration()
+    {
+        // since the CRAM-MD5 provider is registered as part of the JDK, we do not
+        // return the factory class here since we do not need to register it ourselves.
+        if (_hashDirection == HashDirection.PASSWORD_FILE)
+        {
+            return null;
+        }
+        else
+        {
+            //fixme we need a server that will correctly has the incomming plain text for comparison to file.
+            _logger.warn("we need a server that will correctly convert the incomming plain text for comparison to file.");
+            return null;
+        }
+    }
+
+    public void initialise(PrincipalDatabase passwordFile)
+    {
+        initialise(passwordFile, HashDirection.PASSWORD_FILE);
+    }
+
+    public void initialise(PrincipalDatabase passwordFile, HashDirection direction)
+    {
+        super.initialise(passwordFile);
+
+        _hashDirection = direction;
+    }
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainInitialiser.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainInitialiser.java
new file mode 100644
index 0000000000..1d16cd8755
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainInitialiser.java
@@ -0,0 +1,38 @@
+/*
+ *
+ * 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.plain;
+
+import javax.security.sasl.SaslServerFactory;
+
+import org.apache.qpid.server.security.auth.sasl.UsernamePasswordInitialiser;
+
+public class PlainInitialiser extends UsernamePasswordInitialiser
+{
+    public String getMechanismName()
+    {
+        return "PLAIN";
+    }
+
+    public Class<? extends SaslServerFactory> getServerFactoryClassForJCARegistration()
+    {
+        return PlainSaslServerFactory.class;
+    }    
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainSaslServer.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainSaslServer.java
new file mode 100644
index 0000000000..45fb9a4e42
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainSaslServer.java
@@ -0,0 +1,151 @@
+/*
+ *
+ * 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.plain;
+
+import java.io.IOException;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.sasl.AuthorizeCallback;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+
+public class PlainSaslServer implements SaslServer
+{
+    public static final String MECHANISM = "PLAIN";
+
+    private CallbackHandler _cbh;
+
+    private String _authorizationId;
+
+    private boolean _complete = false;
+
+    public PlainSaslServer(CallbackHandler cbh)
+    {
+        _cbh = cbh;
+    }
+
+    public String getMechanismName()
+    {
+        return MECHANISM;
+    }
+
+    public byte[] evaluateResponse(byte[] response) throws SaslException
+    {
+        try
+        {
+            int authzidNullPosition = findNullPosition(response, 0);
+            if (authzidNullPosition < 0)
+            {
+                throw new SaslException("Invalid PLAIN encoding, authzid null terminator not found");
+            }
+            int authcidNullPosition = findNullPosition(response, authzidNullPosition + 1);
+            if (authcidNullPosition < 0)
+            {
+                throw new SaslException("Invalid PLAIN encoding, authcid null terminator not found");
+            }
+
+            // we do not currently support authcid in any meaningful way
+            // String authcid = new String(response, 0, authzidNullPosition, "utf8");
+            String authzid = new String(response, authzidNullPosition + 1, authcidNullPosition - 1, "utf8");
+
+            // we do not care about the prompt but it throws if null
+            NameCallback nameCb = new NameCallback("prompt", authzid);
+            PasswordCallback passwordCb = new PasswordCallback("prompt", false);
+            // TODO: should not get pwd as a String but as a char array...
+            int passwordLen = response.length - authcidNullPosition - 1;
+            String pwd = new String(response, authcidNullPosition + 1, passwordLen, "utf8");
+            AuthorizeCallback authzCb = new AuthorizeCallback(authzid, authzid);
+            Callback[] callbacks = new Callback[]{nameCb, passwordCb, authzCb};
+            _cbh.handle(callbacks);
+            String storedPwd = new String(passwordCb.getPassword());
+            if (storedPwd.equals(pwd))
+            {
+                _complete = true;
+            }
+            if (authzCb.isAuthorized() && _complete)
+            {
+                _authorizationId = authzCb.getAuthenticationID();
+                return null;
+            }
+            else
+            {
+                throw new SaslException("Authentication failed");
+            }
+        }
+        catch (IOException e)
+        {
+            throw new SaslException("Error processing data: " + e, e);
+        }
+        catch (UnsupportedCallbackException e)
+        {
+            throw new SaslException("Unable to obtain data from callback handler: " + e, e);
+        }
+    }
+
+    private int findNullPosition(byte[] response, int startPosition)
+    {
+        int position = startPosition;
+        while (position < response.length)
+        {
+            if (response[position] == (byte) 0)
+            {
+                return position;
+            }
+            position++;
+        }
+        return -1;
+    }
+
+    public boolean isComplete()
+    {
+        return _complete;
+    }
+
+    public String getAuthorizationID()
+    {
+        return _authorizationId;
+    }
+
+    public byte[] unwrap(byte[] incoming, int offset, int len) throws SaslException
+    {
+        throw new SaslException("Unsupported operation");
+    }
+
+    public byte[] wrap(byte[] outgoing, int offset, int len) throws SaslException
+    {
+        throw new SaslException("Unsupported operation");
+    }
+
+    public Object getNegotiatedProperty(String propName)
+    {
+        return null;
+    }
+
+    public void dispose() throws SaslException
+    {
+        _cbh = null;
+    }
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainSaslServerFactory.java b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainSaslServerFactory.java
new file mode 100644
index 0000000000..f0dd9eeb6d
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/plain/PlainSaslServerFactory.java
@@ -0,0 +1,60 @@
+/*
+ *
+ * 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.plain;
+
+import java.util.Map;
+
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import javax.security.sasl.SaslServerFactory;
+
+public class PlainSaslServerFactory implements SaslServerFactory
+{
+    public SaslServer createSaslServer(String mechanism, String protocol, String serverName, Map props,
+                                       CallbackHandler cbh) throws SaslException
+    {
+        if (PlainSaslServer.MECHANISM.equals(mechanism))
+        {
+            return new PlainSaslServer(cbh);
+        }
+        else
+        {
+            return null;
+        }
+    }
+
+    public String[] getMechanismNames(Map props)
+    {
+        if (props.containsKey(Sasl.POLICY_NOPLAINTEXT) ||
+            props.containsKey(Sasl.POLICY_NODICTIONARY) ||
+            props.containsKey(Sasl.POLICY_NOACTIVE))
+        {
+            // returned array must be non null according to interface documentation
+            return new String[0];
+        }
+        else
+        {
+            return new String[]{PlainSaslServer.MECHANISM};
+        }
+    }
+}