merge default into bug23593

27 files changed, 930 insertions, 491 deletions

diff --git a/docs/rabbitmqctl.1.xml b/docs/rabbitmqctl.1.xml
index 01ddd4c1..2152cab3 100644
--- a/docs/rabbitmqctl.1.xml
+++ b/docs/rabbitmqctl.1.xml
@@ -398,7 +398,12 @@
 
     <refsect2>
       <title>User management</title>
-
+      <para>
+        Note that <command>rabbitmqctl</command> manages the RabbitMQ
+        internal user database. Users from any alternative
+        authentication backend will not be visible
+        to <command>rabbitmqctl</command>.
+      </para>
       <variablelist>
         <varlistentry>
           <term><cmdsynopsis><command>add_user</command> <arg choice="req"><replaceable>username</replaceable></arg> <arg choice="req"><replaceable>password</replaceable></arg></cmdsynopsis></term>
@@ -544,7 +549,12 @@
 
     <refsect2>
       <title>Access control</title>
-
+      <para>
+        Note that <command>rabbitmqctl</command> manages the RabbitMQ
+        internal user database. Permissions for users from any
+        alternative authorisation backend will not be visible
+        to <command>rabbitmqctl</command>.
+      </para>
       <variablelist>
         <varlistentry>
           <term><cmdsynopsis><command>add_vhost</command> <arg choice="req"><replaceable>vhostpath</replaceable></arg></cmdsynopsis></term>
@@ -1000,6 +1010,26 @@
                 connection is secured with SSL.</para></listitem>
               </varlistentry>
               <varlistentry>
+                <term>ssl_protocol</term>
+                <listitem><para>SSL protocol
+                (e.g. tlsv1)</para></listitem>
+              </varlistentry>
+              <varlistentry>
+                <term>ssl_key_exchange</term>
+                <listitem><para>SSL key exchange algorithm
+                (e.g. rsa)</para></listitem>
+              </varlistentry>
+              <varlistentry>
+                <term>ssl_cipher</term>
+                <listitem><para>SSL cipher algorithm
+                (e.g. aes_256_cbc)</para></listitem>
+              </varlistentry>
+              <varlistentry>
+                <term>ssl_hash</term>
+                <listitem><para>SSL hash function
+                (e.g. sha)</para></listitem>
+              </varlistentry>
+              <varlistentry>
                 <term>peer_cert_subject</term>
                 <listitem><para>The subject of the peer's SSL
                 certificate, in RFC4514 form.</para></listitem>
diff --git a/ebin/rabbit_app.in b/ebin/rabbit_app.in
index 25d630c0..5ed872b6 100644
--- a/ebin/rabbit_app.in
+++ b/ebin/rabbit_app.in
@@ -32,4 +32,5 @@
          {server_properties, []},
          {collect_statistics, none},
          {auth_mechanisms, ['PLAIN', 'AMQPLAIN']},
+         {auth_backends, [rabbit_auth_backend_internal]},
          {delegate_count, 16}]}]}.
diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index 8c8e12a1..81c3996b 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -29,7 +29,13 @@
 %%   Contributor(s): ______________________________________.
 %%
 
--record(user, {username, password_hash, is_admin}).
+-record(user, {username,
+               is_admin,
+               auth_backend, %% Module this user came from
+               impl          %% Scratch space for that module
+              }).
+
+-record(internal_user, {username, password_hash, is_admin}).
 -record(permission, {configure, write, read}).
 -record(user_vhost, {username, virtual_host}).
 -record(user_permission, {user_vhost, permission}).
diff --git a/include/rabbit_auth_backend_spec.hrl b/include/rabbit_auth_backend_spec.hrl
new file mode 100644
index 00000000..a96c18d8
--- /dev/null
+++ b/include/rabbit_auth_backend_spec.hrl
@@ -0,0 +1,46 @@
+%%   The contents of this file are subject to the Mozilla Public License
+%%   Version 1.1 (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.mozilla.org/MPL/
+%%
+%%   Software distributed under the License is distributed on an "AS IS"
+%%   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+%%   License for the specific language governing rights and limitations
+%%   under the License.
+%%
+%%   The Original Code is RabbitMQ.
+%%
+%%   The Initial Developers of the Original Code are LShift Ltd,
+%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+%%   Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
+%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
+%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%%   (C) 2007-2010 Rabbit Technologies Ltd.
+%%
+%%   All Rights Reserved.
+%%
+%%   Contributor(s): ______________________________________.
+%%
+-ifdef(use_specs).
+
+-spec(description/0 :: () -> [{atom(), any()}]).
+
+-spec(check_user_login/2 :: (rabbit_types:username(), [term()]) ->
+                                 {'ok', rabbit_types:user()} |
+                                 {'refused', string(), [any()]} |
+                                 {'error', any()}).
+-spec(check_vhost_access/3 :: (rabbit_types:user(), rabbit_types:vhost(),
+                               rabbit_access_control:vhost_permission_atom()) ->
+                                   boolean() | {'error', any()}).
+-spec(check_resource_access/3 :: (rabbit_types:user(),
+                                  rabbit_types:r(atom()),
+                                  rabbit_access_control:permission_atom()) ->
+                                      boolean() | {'error', any()}).
+-endif.
diff --git a/scripts/rabbitmq-multi.bat b/scripts/rabbitmq-multi.bat
index a4f8c8b4..ec61dc99 100644
--- a/scripts/rabbitmq-multi.bat
+++ b/scripts/rabbitmq-multi.bat
@@ -89,7 +89,7 @@ if not exist "!ERLANG_HOME!\bin\erl.exe" (
 -pa "!TDP0!..\ebin" ^

 -noinput -hidden ^

 !RABBITMQ_MULTI_ERL_ARGS! ^

--sname rabbitmq_multi ^

+-sname rabbitmq_multi!RANDOM! ^

 !RABBITMQ_CONFIG_ARG! ^

 -s rabbit_multi ^

 !RABBITMQ_MULTI_START_ARGS! ^

diff --git a/scripts/rabbitmq-server.bat b/scripts/rabbitmq-server.bat
index 52a250c6..ec5b4d85 100644
--- a/scripts/rabbitmq-server.bat
+++ b/scripts/rabbitmq-server.bat
@@ -118,7 +118,7 @@ set RABBITMQ_EBIN_ROOT=!TDP0!..\ebin
 -pa "!RABBITMQ_EBIN_ROOT!" ^

 -noinput -hidden ^

 -s rabbit_prelaunch ^

--sname rabbitmqprelaunch%RANDOM% ^

+-sname rabbitmqprelaunch!RANDOM! ^

 -extra "!RABBITMQ_PLUGINS_DIR:\=/!" ^

        "!RABBITMQ_PLUGINS_EXPAND_DIR:\=/!" ^

        "!RABBITMQ_NODENAME!"

diff --git a/scripts/rabbitmqctl.bat b/scripts/rabbitmqctl.bat
index 563b9e58..4ffde73f 100644
--- a/scripts/rabbitmqctl.bat
+++ b/scripts/rabbitmqctl.bat
@@ -58,7 +58,7 @@ if not exist "!ERLANG_HOME!\bin\erl.exe" (
     exit /B

 )

 

-"!ERLANG_HOME!\bin\erl.exe" -pa "!TDP0!..\ebin" -noinput -hidden !RABBITMQ_CTL_ERL_ARGS! -sname rabbitmqctl -s rabbit_control -nodename !RABBITMQ_NODENAME! -extra !STAR!

+"!ERLANG_HOME!\bin\erl.exe" -pa "!TDP0!..\ebin" -noinput -hidden !RABBITMQ_CTL_ERL_ARGS! -sname rabbitmqctl!RANDOM! -s rabbit_control -nodename !RABBITMQ_NODENAME! -extra !STAR!

 

 endlocal

 endlocal

diff --git a/src/rabbit.erl b/src/rabbit.erl
index 2ebfdecf..954e289b 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -458,16 +458,16 @@ insert_default_data() ->
     {ok, DefaultVHost} = application:get_env(default_vhost),
     {ok, [DefaultConfigurePerm, DefaultWritePerm, DefaultReadPerm]} =
         application:get_env(default_permissions),
-    ok = rabbit_access_control:add_vhost(DefaultVHost),
-    ok = rabbit_access_control:add_user(DefaultUser, DefaultPass),
+    ok = rabbit_vhost:add(DefaultVHost),
+    ok = rabbit_auth_backend_internal:add_user(DefaultUser, DefaultPass),
     case DefaultAdmin of
-        true -> rabbit_access_control:set_admin(DefaultUser);
+        true -> rabbit_auth_backend_internal:set_admin(DefaultUser);
         _    -> ok
     end,
-    ok = rabbit_access_control:set_permissions(DefaultUser, DefaultVHost,
-                                               DefaultConfigurePerm,
-                                               DefaultWritePerm,
-                                               DefaultReadPerm),
+    ok = rabbit_auth_backend_internal:set_permissions(DefaultUser, DefaultVHost,
+                                                      DefaultConfigurePerm,
+                                                      DefaultWritePerm,
+                                                      DefaultReadPerm),
     ok.
 
 rotate_logs(File, Suffix, Handler) ->
diff --git a/src/rabbit_access_control.erl b/src/rabbit_access_control.erl
index 51adbac8..02a65442 100644
--- a/src/rabbit_access_control.erl
+++ b/src/rabbit_access_control.erl
@@ -30,70 +30,35 @@
 %%
 
 -module(rabbit_access_control).
--include_lib("stdlib/include/qlc.hrl").
+
 -include("rabbit.hrl").
 
--export([user_pass_login/2, check_user_pass_login/2, make_salt/0,
-         check_vhost_access/2, check_resource_access/3]).
--export([add_user/2, delete_user/1, change_password/2, set_admin/1,
-         clear_admin/1, list_users/0, lookup_user/1, clear_password/1]).
--export([change_password_hash/2, hash_password/1]).
--export([add_vhost/1, delete_vhost/1, vhost_exists/1, list_vhosts/0]).
--export([set_permissions/5, clear_permissions/2,
-         list_permissions/0, list_vhost_permissions/1, list_user_permissions/1,
-         list_user_vhost_permissions/2]).
+-export([user_pass_login/2, check_user_pass_login/2, check_user_login/2,
+         check_vhost_access/2, check_resource_access/3, list_vhosts/2]).
 
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
 
--export_type([username/0, password/0, password_hash/0]).
+-export_type([permission_atom/0, vhost_permission_atom/0]).
 
 -type(permission_atom() :: 'configure' | 'read' | 'write').
--type(username() :: binary()).
--type(password() :: binary()).
--type(password_hash() :: binary()).
--type(regexp() :: binary()).
+-type(vhost_permission_atom() :: 'read' | 'write').
+
 -spec(user_pass_login/2 ::
-        (username(), password())
+        (rabbit_types:username(), rabbit_types:password())
         -> rabbit_types:user() | rabbit_types:channel_exit()).
 -spec(check_user_pass_login/2 ::
-        (username(), password())
+        (rabbit_types:username(), rabbit_types:password())
         -> {'ok', rabbit_types:user()} | {'refused', string(), [any()]}).
--spec(make_salt/0 :: () -> binary()).
 -spec(check_vhost_access/2 ::
         (rabbit_types:user(), rabbit_types:vhost())
         -> 'ok' | rabbit_types:channel_exit()).
 -spec(check_resource_access/3 ::
-        (username(), rabbit_types:r(atom()), permission_atom())
+        (rabbit_types:user(), rabbit_types:r(atom()), permission_atom())
         -> 'ok' | rabbit_types:channel_exit()).
--spec(add_user/2 :: (username(), password()) -> 'ok').
--spec(delete_user/1 :: (username()) -> 'ok').
--spec(change_password/2 :: (username(), password()) -> 'ok').
--spec(clear_password/1 :: (username()) -> 'ok').
--spec(change_password_hash/2 :: (username(), password_hash()) -> 'ok').
--spec(hash_password/1 :: (password()) -> password_hash()).
--spec(set_admin/1 :: (username()) -> 'ok').
--spec(clear_admin/1 :: (username()) -> 'ok').
--spec(list_users/0 :: () -> [{username(), boolean()}]).
--spec(lookup_user/1 ::
-        (username()) -> rabbit_types:ok(rabbit_types:user())
-                            | rabbit_types:error('not_found')).
--spec(add_vhost/1 :: (rabbit_types:vhost()) -> 'ok').
--spec(delete_vhost/1 :: (rabbit_types:vhost()) -> 'ok').
--spec(vhost_exists/1 :: (rabbit_types:vhost()) -> boolean()).
--spec(list_vhosts/0 :: () -> [rabbit_types:vhost()]).
--spec(set_permissions/5 ::(username(), rabbit_types:vhost(), regexp(),
-                           regexp(), regexp()) -> 'ok').
--spec(clear_permissions/2 :: (username(), rabbit_types:vhost()) -> 'ok').
--spec(list_permissions/0 ::
-        () -> [{username(), rabbit_types:vhost(), regexp(), regexp(), regexp()}]).
--spec(list_vhost_permissions/1 ::
-        (rabbit_types:vhost()) -> [{username(), regexp(), regexp(), regexp()}]).
--spec(list_user_permissions/1 ::
-        (username()) -> [{rabbit_types:vhost(), regexp(), regexp(), regexp()}]).
--spec(list_user_vhost_permissions/2 ::
-        (username(), rabbit_types:vhost()) -> [{regexp(), regexp(), regexp()}]).
+-spec(list_vhosts/2 :: (rabbit_types:user(), vhost_permission_atom())
+                       -> [rabbit_types:vhost()]).
 
 -endif.
 
@@ -109,314 +74,79 @@ user_pass_login(User, Pass) ->
             U
     end.
 
-check_user_pass_login(Username, Pass) ->
-    Refused = {refused, "user '~s' - invalid credentials", [Username]},
-    case lookup_user(Username) of
-        {ok, User} ->
-            case check_password(Pass, User#user.password_hash) of
-                true -> {ok, User};
-                _    -> Refused
-            end;
-        {error, not_found} ->
-            Refused
-    end.
-
-internal_lookup_vhost_access(Username, VHostPath) ->
-    %% TODO: use dirty ops instead
-    rabbit_misc:execute_mnesia_transaction(
-      fun () ->
-              case mnesia:read({rabbit_user_permission,
-                                #user_vhost{username     = Username,
-                                            virtual_host = VHostPath}}) of
-                  [] -> not_found;
-                  [R] -> {ok, R}
-              end
-      end).
-
-check_vhost_access(#user{username = Username}, VHostPath) ->
+check_user_pass_login(Username, Password) ->
+    check_user_login(Username, [{password, Password}]).
+
+check_user_login(Username, AuthProps) ->
+    {ok, Modules} = application:get_env(rabbit, auth_backends),
+    lists:foldl(
+      fun(Module, {refused, _, _}) ->
+              case Module:check_user_login(Username, AuthProps) of
+                  {error, E} ->
+                      {refused, "~s failed authenticating ~s: ~p~n",
+                       [Module, Username, E]};
+                  Else ->
+                      Else
+              end;
+         (_, {ok, User}) ->
+              {ok, User}
+      end, {refused, "No modules checked '~s'", [Username]}, Modules).
+
+check_vhost_access(User = #user{ username     = Username,
+                                 auth_backend = Module }, VHostPath) ->
     ?LOGDEBUG("Checking VHost access for ~p to ~p~n", [Username, VHostPath]),
-    case internal_lookup_vhost_access(Username, VHostPath) of
-        {ok, _R} ->
-            ok;
-        not_found ->
-            rabbit_misc:protocol_error(
-              access_refused, "access to vhost '~s' refused for user '~s'",
-              [VHostPath, Username])
-    end.
-
-permission_index(configure) -> #permission.configure;
-permission_index(write)     -> #permission.write;
-permission_index(read)      -> #permission.read.
-
-check_resource_access(Username,
-                      R = #resource{kind = exchange, name = <<"">>},
+    check_access(
+      fun() ->
+              rabbit_vhost:exists(VHostPath) andalso
+                  Module:check_vhost_access(User, VHostPath, write)
+      end,
+      "~s failed checking vhost access to ~s for ~s: ~p~n",
+      [Module, VHostPath, Username],
+      "access to vhost '~s' refused for user '~s'",
+      [VHostPath, Username]).
+
+check_resource_access(User, R = #resource{kind = exchange, name = <<"">>},
                       Permission) ->
-    check_resource_access(Username,
-                          R#resource{name = <<"amq.default">>},
+    check_resource_access(User, R#resource{name = <<"amq.default">>},
                           Permission);
-check_resource_access(Username,
-                      R = #resource{virtual_host = VHostPath, name = Name},
-                      Permission) ->
-    Res = case mnesia:dirty_read({rabbit_user_permission,
-                                  #user_vhost{username     = Username,
-                                              virtual_host = VHostPath}}) of
-              [] ->
-                  false;
-              [#user_permission{permission = P}] ->
-                  PermRegexp =
-                      case element(permission_index(Permission), P) of
-                          %% <<"^$">> breaks Emacs' erlang mode
-                          <<"">> -> <<$^, $$>>;
-                          RE     -> RE
-                      end,
-                  case re:run(Name, PermRegexp, [{capture, none}]) of
-                      match    -> true;
-                      nomatch  -> false
-                  end
-          end,
-    if Res  -> ok;
-       true -> rabbit_misc:protocol_error(
-                 access_refused, "access to ~s refused for user '~s'",
-                 [rabbit_misc:rs(R), Username])
-    end.
-
-add_user(Username, Password) ->
-    R = rabbit_misc:execute_mnesia_transaction(
-          fun () ->
-                  case mnesia:wread({rabbit_user, Username}) of
-                      [] ->
-                          ok = mnesia:write(rabbit_user,
-                                            #user{username = Username,
-                                                  password_hash =
-                                                      hash_password(Password),
-                                                  is_admin = false},
-                                            write);
-                      _ ->
-                          mnesia:abort({user_already_exists, Username})
-                  end
-          end),
-    rabbit_log:info("Created user ~p~n", [Username]),
-    R.
-
-delete_user(Username) ->
-    R = rabbit_misc:execute_mnesia_transaction(
-          rabbit_misc:with_user(
-            Username,
-            fun () ->
-                    ok = mnesia:delete({rabbit_user, Username}),
-                    [ok = mnesia:delete_object(
-                            rabbit_user_permission, R, write) ||
-                        R <- mnesia:match_object(
-                               rabbit_user_permission,
-                               #user_permission{user_vhost = #user_vhost{
-                                                  username = Username,
-                                                  virtual_host = '_'},
-                                                permission = '_'},
-                               write)],
-                    ok
-            end)),
-    rabbit_log:info("Deleted user ~p~n", [Username]),
-    R.
-
-change_password(Username, Password) ->
-    change_password_hash(Username, hash_password(Password)).
-
-clear_password(Username) ->
-    change_password_hash(Username, <<"">>).
-
-change_password_hash(Username, PasswordHash) ->
-    R = update_user(Username, fun(User) ->
-                                      User#user{ password_hash = PasswordHash }
-                              end),
-    rabbit_log:info("Changed password for user ~p~n", [Username]),
-    R.
-
-hash_password(Cleartext) ->
-    Salt = make_salt(),
-    Hash = salted_md5(Salt, Cleartext),
-    <<Salt/binary, Hash/binary>>.
-
-check_password(Cleartext, <<Salt:4/binary, Hash/binary>>) ->
-    Hash =:= salted_md5(Salt, Cleartext).
-
-make_salt() ->
-    {A1,A2,A3} = now(),
-    random:seed(A1, A2, A3),
-    Salt = random:uniform(16#ffffffff),
-    <<Salt:32>>.
-
-salted_md5(Salt, Cleartext) ->
-    Salted = <<Salt/binary, Cleartext/binary>>,
-    erlang:md5(Salted).
-
-set_admin(Username) ->
-    set_admin(Username, true).
-
-clear_admin(Username) ->
-    set_admin(Username, false).
-
-set_admin(Username, IsAdmin) ->
-    R = update_user(Username, fun(User) ->
-                                      User#user{is_admin = IsAdmin}
-                              end),
-    rabbit_log:info("Set user admin flag for user ~p to ~p~n",
-                    [Username, IsAdmin]),
-    R.
-
-update_user(Username, Fun) ->
-    rabbit_misc:execute_mnesia_transaction(
-      rabbit_misc:with_user(
-        Username,
-        fun () ->
-                {ok, User} = lookup_user(Username),
-                ok = mnesia:write(rabbit_user, Fun(User), write)
-        end)).
-
-list_users() ->
-    [{Username, IsAdmin} ||
-        #user{username = Username, is_admin = IsAdmin} <-
-            mnesia:dirty_match_object(rabbit_user, #user{_ = '_'})].
-
-lookup_user(Username) ->
-    rabbit_misc:dirty_read({rabbit_user, Username}).
-
-add_vhost(VHostPath) ->
-    R = rabbit_misc:execute_mnesia_transaction(
-          fun () ->
-                  case mnesia:wread({rabbit_vhost, VHostPath}) of
-                      [] ->
-                          ok = mnesia:write(rabbit_vhost,
-                                            #vhost{virtual_host = VHostPath},
-                                            write),
-                          [rabbit_exchange:declare(
-                             rabbit_misc:r(VHostPath, exchange, Name),
-                             Type, true, false, false, []) ||
-                              {Name,Type} <-
-                                  [{<<"">>,           direct},
-                                   {<<"amq.direct">>, direct},
-                                   {<<"amq.topic">>,  topic},
-                                   {<<"amq.match">>,  headers}, %% per 0-9-1 pdf
-                                   {<<"amq.headers">>,  headers}, %% per 0-9-1 xml
-                                   {<<"amq.fanout">>, fanout}]],
-                          ok;
-                      [_] ->
-                          mnesia:abort({vhost_already_exists, VHostPath})
-                  end
-          end),
-    rabbit_log:info("Added vhost ~p~n", [VHostPath]),
-    R.
-
-delete_vhost(VHostPath) ->
-    %%FIXME: We are forced to delete the queues outside the TX below
-    %%because queue deletion involves sending messages to the queue
-    %%process, which in turn results in further mnesia actions and
-    %%eventually the termination of that process.
-    lists:foreach(fun (Q) ->
-                          {ok,_} = rabbit_amqqueue:delete(Q, false, false)
-                  end,
-                  rabbit_amqqueue:list(VHostPath)),
-    R = rabbit_misc:execute_mnesia_transaction(
-          rabbit_misc:with_vhost(
-            VHostPath,
-            fun () ->
-                    ok = internal_delete_vhost(VHostPath)
-            end)),
-    rabbit_log:info("Deleted vhost ~p~n", [VHostPath]),
-    R.
-
-internal_delete_vhost(VHostPath) ->
-    lists:foreach(fun (#exchange{name = Name}) ->
-                          ok = rabbit_exchange:delete(Name, false)
-                  end,
-                  rabbit_exchange:list(VHostPath)),
-    lists:foreach(fun ({Username, _, _, _}) ->
-                          ok = clear_permissions(Username, VHostPath)
-                  end,
-                  list_vhost_permissions(VHostPath)),
-    ok = mnesia:delete({rabbit_vhost, VHostPath}),
-    ok.
-
-vhost_exists(VHostPath) ->
-    mnesia:dirty_read({rabbit_vhost, VHostPath}) /= [].
-
-list_vhosts() ->
-    mnesia:dirty_all_keys(rabbit_vhost).
-
-validate_regexp(RegexpBin) ->
-    Regexp = binary_to_list(RegexpBin),
-    case re:compile(Regexp) of
-        {ok, _}         -> ok;
-        {error, Reason} -> throw({error, {invalid_regexp, Regexp, Reason}})
+check_resource_access(User = #user{username = Username, auth_backend = Module},
+                      Resource, Permission) ->
+    check_access(
+      fun() -> Module:check_resource_access(User, Resource, Permission) end,
+      "~s failed checking resource access to ~p for ~s: ~p~n",
+      [Module, Resource, Username],
+      "access to ~s refused for user '~s'",
+      [rabbit_misc:rs(Resource), Username]).
+
+check_access(Fun, ErrStr, ErrArgs, RefStr, RefArgs) ->
+    Allow = case Fun() of
+                {error, _} = E ->
+                    rabbit_log:error(ErrStr, ErrArgs ++ [E]),
+                    false;
+                Else ->
+                    Else
+            end,
+    case Allow of
+        true ->
+            ok;
+        false ->
+            rabbit_misc:protocol_error(access_refused, RefStr, RefArgs)
     end.
 
-set_permissions(Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm) ->
-    lists:map(fun validate_regexp/1, [ConfigurePerm, WritePerm, ReadPerm]),
-    rabbit_misc:execute_mnesia_transaction(
-      rabbit_misc:with_user_and_vhost(
-        Username, VHostPath,
-        fun () -> ok = mnesia:write(
-                         rabbit_user_permission,
-                         #user_permission{user_vhost = #user_vhost{
-                                            username     = Username,
-                                            virtual_host = VHostPath},
-                                          permission = #permission{
-                                            configure = ConfigurePerm,
-                                            write     = WritePerm,
-                                            read      = ReadPerm}},
-                         write)
-        end)).
-
-
-clear_permissions(Username, VHostPath) ->
-    rabbit_misc:execute_mnesia_transaction(
-      rabbit_misc:with_user_and_vhost(
-        Username, VHostPath,
-        fun () ->
-                ok = mnesia:delete({rabbit_user_permission,
-                                    #user_vhost{username     = Username,
-                                                virtual_host = VHostPath}})
-        end)).
-
-list_permissions() ->
-    [{Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm} ||
-        {Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm} <-
-            list_permissions(match_user_vhost('_', '_'))].
-
-list_vhost_permissions(VHostPath) ->
-    [{Username, ConfigurePerm, WritePerm, ReadPerm} ||
-        {Username, _, ConfigurePerm, WritePerm, ReadPerm} <-
-            list_permissions(rabbit_misc:with_vhost(
-                               VHostPath, match_user_vhost('_', VHostPath)))].
-
-list_user_permissions(Username) ->
-    [{VHostPath, ConfigurePerm, WritePerm, ReadPerm} ||
-        {_, VHostPath, ConfigurePerm, WritePerm, ReadPerm} <-
-            list_permissions(rabbit_misc:with_user(
-                               Username, match_user_vhost(Username, '_')))].
-
-list_user_vhost_permissions(Username, VHostPath) ->
-    [{ConfigurePerm, WritePerm, ReadPerm} ||
-        {_, _, ConfigurePerm, WritePerm, ReadPerm} <-
-            list_permissions(rabbit_misc:with_user_and_vhost(
-                               Username, VHostPath,
-                               match_user_vhost(Username, VHostPath)))].
-
-list_permissions(QueryThunk) ->
-    [{Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm} ||
-        #user_permission{user_vhost = #user_vhost{username     = Username,
-                                                  virtual_host = VHostPath},
-                         permission = #permission{ configure = ConfigurePerm,
-                                                   write     = WritePerm,
-                                                   read      = ReadPerm}} <-
-            %% TODO: use dirty ops instead
-            rabbit_misc:execute_mnesia_transaction(QueryThunk)].
-
-match_user_vhost(Username, VHostPath) ->
-    fun () -> mnesia:match_object(
-                rabbit_user_permission,
-                #user_permission{user_vhost = #user_vhost{
-                                   username     = Username,
-                                   virtual_host = VHostPath},
-                                 permission = '_'},
-                read)
-    end.
+%% Permission = write -> log in
+%% Permission = read  -> learn of the existence of (only relevant for
+%%                       management plugin)
+list_vhosts(User = #user{username = Username, auth_backend = Module},
+            Permission) ->
+    lists:filter(
+      fun(VHost) ->
+              case Module:check_vhost_access(User, VHost, Permission) of
+                  {error, _} = E ->
+                      rabbit_log:warning("~w failed checking vhost access "
+                                         "to ~s for ~s: ~p~n",
+                                         [Module, VHost, Username, E]),
+                      false;
+                  Else ->
+                      Else
+              end
+      end, rabbit_vhost:list()).
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index cc1e50f7..f2affd9f 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -455,7 +455,7 @@ group_confirms_by_channel([{Ch, Msg1} | CMs], Acc) ->
     group_confirms_by_channel(CMs, [{Ch, [Msg1]} | Acc]).
 
 record_confirm_message(#delivery{msg_seq_no = undefined}, State) ->
-    State;
+    {no_confirm, State};
 record_confirm_message(#delivery{sender     = ChPid,
                                  msg_seq_no = MsgSeqNo,
                                  message    = #basic_message {
@@ -464,9 +464,10 @@ record_confirm_message(#delivery{sender     = ChPid,
                        State =
                            #q{guid_to_channel = GTC,
                               q               = #amqqueue{durable = true}}) ->
-    State#q{guid_to_channel = dict:store(Guid, {ChPid, MsgSeqNo}, GTC)};
+    {confirm,
+     State#q{guid_to_channel = dict:store(Guid, {ChPid, MsgSeqNo}, GTC)}};
 record_confirm_message(_Delivery, State) ->
-    State.
+    {no_confirm, State}.
 
 run_message_queue(State) ->
     Funs = {fun deliver_from_queue_pred/2,
@@ -481,12 +482,12 @@ attempt_delivery(#delivery{txn        = none,
                            sender     = ChPid,
                            message    = Message,
                            msg_seq_no = MsgSeqNo},
-                 State = #q{backing_queue = BQ, q = Q}) ->
-    NeedsConfirming = Message#basic_message.is_persistent andalso
-                      Q#amqqueue.durable,
-    case NeedsConfirming of
-        false -> rabbit_channel:confirm(ChPid, [MsgSeqNo]);
-        _     -> ok
+                 {NeedsConfirming, State = #q{backing_queue = BQ}}) ->
+    %% must confirm immediately if it has a MsgSeqNo and not NeedsConfirming
+    case {NeedsConfirming, MsgSeqNo} of
+        {_, undefined}  -> ok;
+        {no_confirm, _} -> rabbit_channel:confirm(ChPid, [MsgSeqNo]);
+        {confirm, _}    -> ok
     end,
     PredFun = fun (IsEmpty, _State) -> not IsEmpty end,
     DeliverFun =
@@ -498,31 +499,37 @@ attempt_delivery(#delivery{txn        = none,
                     BQ:publish_delivered(
                       AckRequired, Message,
                       (?BASE_MESSAGE_PROPERTIES)#message_properties{
-                        needs_confirming = NeedsConfirming},
+                        needs_confirming = (NeedsConfirming =:= confirm)},
                       BQS),
                 {{Message, false, AckTag}, true,
                  State1#q{backing_queue_state = BQS1}}
         end,
-    deliver_msgs_to_consumers({ PredFun, DeliverFun }, false, State);
+    {Delivered, State1} =
+        deliver_msgs_to_consumers({ PredFun, DeliverFun }, false, State),
+    {Delivered, NeedsConfirming, State1};
 attempt_delivery(#delivery{txn = Txn,
                            sender  = ChPid,
                            message = Message},
-                 State = #q{backing_queue = BQ,
-                            backing_queue_state = BQS}) ->
+                 {NeedsConfirming,
+                  State = #q{backing_queue = BQ,
+                            backing_queue_state = BQS}}) ->
     record_current_channel_tx(ChPid, Txn),
     {true,
+     NeedsConfirming,
      State#q{backing_queue_state =
                  BQ:tx_publish(Txn, Message, ?BASE_MESSAGE_PROPERTIES, BQS)}}.
 
 deliver_or_enqueue(Delivery, State) ->
     case attempt_delivery(Delivery, record_confirm_message(Delivery, State)) of
-        {true, State1} ->
+        {true, _, State1} ->
             {true, State1};
-        {false, State1 = #q{backing_queue = BQ, backing_queue_state = BQS}} ->
-            #delivery{message = Message, msg_seq_no = MsgSeqNo} = Delivery,
+        {false, NeedsConfirming, State1 = #q{backing_queue = BQ,
+                                             backing_queue_state = BQS}} ->
+            #delivery{message = Message} = Delivery,
             BQS1 = BQ:publish(Message,
                               (message_properties(State)) #message_properties{
-                                needs_confirming = (MsgSeqNo =/= undefined)},
+                                needs_confirming =
+                                    (NeedsConfirming =:= confirm)},
                               BQS),
             {false, ensure_ttl_timer(State1#q{backing_queue_state = BQS1})}
     end.
@@ -846,7 +853,7 @@ handle_call({deliver_immediately, Delivery},
     %% just all ready-to-consume queues get the message, with unready
     %% queues discarding the message?
     %%
-    {Delivered, State1} =
+    {Delivered, _NeedsConfirming, State1} =
         attempt_delivery(Delivery, record_confirm_message(Delivery, State)),
     reply(Delivered, State1);
 
diff --git a/src/rabbit_auth_backend.erl b/src/rabbit_auth_backend.erl
new file mode 100644
index 00000000..0dc8e61b
--- /dev/null
+++ b/src/rabbit_auth_backend.erl
@@ -0,0 +1,76 @@
+%%   The contents of this file are subject to the Mozilla Public License
+%%   Version 1.1 (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.mozilla.org/MPL/
+%%
+%%   Software distributed under the License is distributed on an "AS IS"
+%%   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+%%   License for the specific language governing rights and limitations
+%%   under the License.
+%%
+%%   The Original Code is RabbitMQ.
+%%
+%%   The Initial Developers of the Original Code are LShift Ltd,
+%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+%%   Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
+%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
+%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%%   (C) 2007-2010 Rabbit Technologies Ltd.
+%%
+%%   All Rights Reserved.
+%%
+%%   Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_auth_backend).
+
+-export([behaviour_info/1]).
+
+behaviour_info(callbacks) ->
+    [
+     %% A description proplist as with auth mechanisms,
+     %% exchanges. Currently unused.
+     {description, 0},
+
+     %% Check a user can log in, given a username and a proplist of
+     %% authentication information (e.g. [{password, Password}]).
+     %%
+     %% Possible responses:
+     %% {ok, User}
+     %%     Authentication succeeded, and here's the user record.
+     %% {error, Error}
+     %%     Something went wrong. Log and die.
+     %% {refused, Msg, Args}
+     %%     Client failed authentication. Log and die.
+     {check_user_login, 2},
+
+     %% Given #user, vhost path and permission, can a user access a vhost?
+     %% Permission is read  - learn of the existence of (only relevant for
+     %%                       management plugin)
+     %%            or write - log in
+     %%
+     %% Possible responses:
+     %% true
+     %% false
+     %% {error, Error}
+     %%     Something went wrong. Log and die.
+     {check_vhost_access, 3},
+
+     %% Given #user, resource and permission, can a user access a resource?
+     %%
+     %% Possible responses:
+     %% true
+     %% false
+     %% {error, Error}
+     %%     Something went wrong. Log and die.
+     {check_resource_access, 3}
+    ];
+behaviour_info(_Other) ->
+    undefined.
diff --git a/src/rabbit_auth_backend_internal.erl b/src/rabbit_auth_backend_internal.erl
new file mode 100644
index 00000000..233e2b90
--- /dev/null
+++ b/src/rabbit_auth_backend_internal.erl
@@ -0,0 +1,347 @@
+%%   The contents of this file are subject to the Mozilla Public License
+%%   Version 1.1 (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.mozilla.org/MPL/
+%%
+%%   Software distributed under the License is distributed on an "AS IS"
+%%   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+%%   License for the specific language governing rights and limitations
+%%   under the License.
+%%
+%%   The Original Code is RabbitMQ.
+%%
+%%   The Initial Developers of the Original Code are LShift Ltd,
+%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+%%   Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
+%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
+%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%%   (C) 2007-2010 Rabbit Technologies Ltd.
+%%
+%%   All Rights Reserved.
+%%
+%%   Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_auth_backend_internal).
+-include("rabbit.hrl").
+
+-behaviour(rabbit_auth_backend).
+
+-export([description/0]).
+-export([check_user_login/2, check_vhost_access/3, check_resource_access/3]).
+
+-export([add_user/2, delete_user/1, change_password/2, set_admin/1,
+         clear_admin/1, list_users/0, lookup_user/1, clear_password/1]).
+-export([make_salt/0, check_password/2, change_password_hash/2,
+         hash_password/1]).
+-export([set_permissions/5, clear_permissions/2,
+         list_permissions/0, list_vhost_permissions/1, list_user_permissions/1,
+         list_user_vhost_permissions/2]).
+
+-include("rabbit_auth_backend_spec.hrl").
+
+-ifdef(use_specs).
+
+-type(regexp() :: binary()).
+
+-spec(add_user/2 :: (rabbit_types:username(), rabbit_types:password()) -> 'ok').
+-spec(delete_user/1 :: (rabbit_types:username()) -> 'ok').
+-spec(change_password/2 :: (rabbit_types:username(), rabbit_types:password())
+                           -> 'ok').
+-spec(clear_password/1 :: (rabbit_types:username()) -> 'ok').
+-spec(make_salt/0 :: () -> binary()).
+-spec(check_password/2 :: (rabbit_types:password(),
+                           rabbit_types:password_hash()) -> boolean()).
+-spec(change_password_hash/2 :: (rabbit_types:username(),
+                                 rabbit_types:password_hash()) -> 'ok').
+-spec(hash_password/1 :: (rabbit_types:password())
+                         -> rabbit_types:password_hash()).
+-spec(set_admin/1 :: (rabbit_types:username()) -> 'ok').
+-spec(clear_admin/1 :: (rabbit_types:username()) -> 'ok').
+-spec(list_users/0 :: () -> [{rabbit_types:username(), boolean()}]).
+-spec(lookup_user/1 :: (rabbit_types:username())
+        -> rabbit_types:ok(rabbit_types:internal_user())
+               | rabbit_types:error('not_found')).
+-spec(set_permissions/5 ::(rabbit_types:username(), rabbit_types:vhost(),
+                           regexp(), regexp(), regexp()) -> 'ok').
+-spec(clear_permissions/2 :: (rabbit_types:username(), rabbit_types:vhost())
+                             -> 'ok').
+-spec(list_permissions/0 ::
+        () -> [{rabbit_types:username(), rabbit_types:vhost(),
+                regexp(), regexp(), regexp()}]).
+-spec(list_vhost_permissions/1 ::
+        (rabbit_types:vhost()) -> [{rabbit_types:username(),
+                                    regexp(), regexp(), regexp()}]).
+-spec(list_user_permissions/1 ::
+        (rabbit_types:username()) -> [{rabbit_types:vhost(),
+                                       regexp(), regexp(), regexp()}]).
+-spec(list_user_vhost_permissions/2 ::
+        (rabbit_types:username(), rabbit_types:vhost())
+        -> [{regexp(), regexp(), regexp()}]).
+
+-endif.
+
+%%----------------------------------------------------------------------------
+
+%% Implementation of rabbit_auth_backend
+
+description() ->
+    [{name, <<"Internal">>},
+     {description, <<"Internal user / password database">>}].
+
+check_user_login(Username, []) ->
+    internal_check_user_login(Username, fun(_) -> true end);
+check_user_login(Username, [{password, Password}]) ->
+    internal_check_user_login(
+      Username,
+      fun(#internal_user{password_hash = Hash}) ->
+              check_password(Password, Hash)
+      end);
+check_user_login(Username, AuthProps) ->
+    exit({unknown_auth_props, Username, AuthProps}).
+
+internal_check_user_login(Username, Fun) ->
+    Refused = {refused, "user '~s' - invalid credentials", [Username]},
+    case lookup_user(Username) of
+        {ok, User = #internal_user{is_admin = IsAdmin}} ->
+            case Fun(User) of
+                true -> {ok, #user{username     = Username,
+                                   is_admin     = IsAdmin,
+                                   auth_backend = ?MODULE,
+                                   impl         = User}};
+                _    -> Refused
+            end;
+        {error, not_found} ->
+            Refused
+    end.
+
+check_vhost_access(#user{is_admin = true},    _VHostPath, read) ->
+    true;
+
+check_vhost_access(#user{username = Username}, VHostPath, _) ->
+    %% TODO: use dirty ops instead
+    rabbit_misc:execute_mnesia_transaction(
+      fun () ->
+              case mnesia:read({rabbit_user_permission,
+                                #user_vhost{username     = Username,
+                                            virtual_host = VHostPath}}) of
+                  []   -> false;
+                  [_R] -> true
+              end
+      end).
+
+check_resource_access(#user{username = Username},
+                      #resource{virtual_host = VHostPath, name = Name},
+                      Permission) ->
+    case mnesia:dirty_read({rabbit_user_permission,
+                            #user_vhost{username     = Username,
+                                        virtual_host = VHostPath}}) of
+        [] ->
+            false;
+        [#user_permission{permission = P}] ->
+            PermRegexp =
+                case element(permission_index(Permission), P) of
+                    %% <<"^$">> breaks Emacs' erlang mode
+                    <<"">> -> <<$^, $$>>;
+                    RE     -> RE
+                end,
+            case re:run(Name, PermRegexp, [{capture, none}]) of
+                match    -> true;
+                nomatch  -> false
+            end
+    end.
+
+permission_index(configure) -> #permission.configure;
+permission_index(write)     -> #permission.write;
+permission_index(read)      -> #permission.read.
+
+%%----------------------------------------------------------------------------
+%% Manipulation of the user database
+
+add_user(Username, Password) ->
+    R = rabbit_misc:execute_mnesia_transaction(
+          fun () ->
+                  case mnesia:wread({rabbit_user, Username}) of
+                      [] ->
+                          ok = mnesia:write(
+                                 rabbit_user,
+                                 #internal_user{username = Username,
+                                                password_hash =
+                                                    hash_password(Password),
+                                                is_admin = false},
+                                 write);
+                      _ ->
+                          mnesia:abort({user_already_exists, Username})
+                  end
+          end),
+    rabbit_log:info("Created user ~p~n", [Username]),
+    R.
+
+delete_user(Username) ->
+    R = rabbit_misc:execute_mnesia_transaction(
+          rabbit_misc:with_user(
+            Username,
+            fun () ->
+                    ok = mnesia:delete({rabbit_user, Username}),
+                    [ok = mnesia:delete_object(
+                            rabbit_user_permission, R, write) ||
+                        R <- mnesia:match_object(
+                               rabbit_user_permission,
+                               #user_permission{user_vhost = #user_vhost{
+                                                  username = Username,
+                                                  virtual_host = '_'},
+                                                permission = '_'},
+                               write)],
+                    ok
+            end)),
+    rabbit_log:info("Deleted user ~p~n", [Username]),
+    R.
+
+change_password(Username, Password) ->
+    change_password_hash(Username, hash_password(Password)).
+
+clear_password(Username) ->
+    change_password_hash(Username, <<"">>).
+
+change_password_hash(Username, PasswordHash) ->
+    R = update_user(Username, fun(User) ->
+                                      User#internal_user{
+                                        password_hash = PasswordHash }
+                              end),
+    rabbit_log:info("Changed password for user ~p~n", [Username]),
+    R.
+
+hash_password(Cleartext) ->
+    Salt = make_salt(),
+    Hash = salted_md5(Salt, Cleartext),
+    <<Salt/binary, Hash/binary>>.
+
+check_password(Cleartext, <<Salt:4/binary, Hash/binary>>) ->
+    Hash =:= salted_md5(Salt, Cleartext).
+
+make_salt() ->
+    {A1,A2,A3} = now(),
+    random:seed(A1, A2, A3),
+    Salt = random:uniform(16#ffffffff),
+    <<Salt:32>>.
+
+salted_md5(Salt, Cleartext) ->
+    Salted = <<Salt/binary, Cleartext/binary>>,
+    erlang:md5(Salted).
+
+set_admin(Username) ->
+    set_admin(Username, true).
+
+clear_admin(Username) ->
+    set_admin(Username, false).
+
+set_admin(Username, IsAdmin) ->
+    R = update_user(Username, fun(User) ->
+                                      User#internal_user{is_admin = IsAdmin}
+                              end),
+    rabbit_log:info("Set user admin flag for user ~p to ~p~n",
+                    [Username, IsAdmin]),
+    R.
+
+update_user(Username, Fun) ->
+    rabbit_misc:execute_mnesia_transaction(
+      rabbit_misc:with_user(
+        Username,
+        fun () ->
+                {ok, User} = lookup_user(Username),
+                ok = mnesia:write(rabbit_user, Fun(User), write)
+        end)).
+
+list_users() ->
+    [{Username, IsAdmin} ||
+        #internal_user{username = Username, is_admin = IsAdmin} <-
+            mnesia:dirty_match_object(rabbit_user, #internal_user{_ = '_'})].
+
+lookup_user(Username) ->
+    rabbit_misc:dirty_read({rabbit_user, Username}).
+
+validate_regexp(RegexpBin) ->
+    Regexp = binary_to_list(RegexpBin),
+    case re:compile(Regexp) of
+        {ok, _}         -> ok;
+        {error, Reason} -> throw({error, {invalid_regexp, Regexp, Reason}})
+    end.
+
+set_permissions(Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm) ->
+    lists:map(fun validate_regexp/1, [ConfigurePerm, WritePerm, ReadPerm]),
+    rabbit_misc:execute_mnesia_transaction(
+      rabbit_misc:with_user_and_vhost(
+        Username, VHostPath,
+        fun () -> ok = mnesia:write(
+                         rabbit_user_permission,
+                         #user_permission{user_vhost = #user_vhost{
+                                            username     = Username,
+                                            virtual_host = VHostPath},
+                                          permission = #permission{
+                                            configure = ConfigurePerm,
+                                            write     = WritePerm,
+                                            read      = ReadPerm}},
+                         write)
+        end)).
+
+
+clear_permissions(Username, VHostPath) ->
+    rabbit_misc:execute_mnesia_transaction(
+      rabbit_misc:with_user_and_vhost(
+        Username, VHostPath,
+        fun () ->
+                ok = mnesia:delete({rabbit_user_permission,
+                                    #user_vhost{username     = Username,
+                                                virtual_host = VHostPath}})
+        end)).
+
+list_permissions() ->
+    [{Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm} ||
+        {Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm} <-
+            list_permissions(match_user_vhost('_', '_'))].
+
+list_vhost_permissions(VHostPath) ->
+    [{Username, ConfigurePerm, WritePerm, ReadPerm} ||
+        {Username, _, ConfigurePerm, WritePerm, ReadPerm} <-
+            list_permissions(rabbit_vhost:with(
+                               VHostPath, match_user_vhost('_', VHostPath)))].
+
+list_user_permissions(Username) ->
+    [{VHostPath, ConfigurePerm, WritePerm, ReadPerm} ||
+        {_, VHostPath, ConfigurePerm, WritePerm, ReadPerm} <-
+            list_permissions(rabbit_misc:with_user(
+                               Username, match_user_vhost(Username, '_')))].
+
+list_user_vhost_permissions(Username, VHostPath) ->
+    [{ConfigurePerm, WritePerm, ReadPerm} ||
+        {_, _, ConfigurePerm, WritePerm, ReadPerm} <-
+            list_permissions(rabbit_misc:with_user_and_vhost(
+                               Username, VHostPath,
+                               match_user_vhost(Username, VHostPath)))].
+
+list_permissions(QueryThunk) ->
+    [{Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm} ||
+        #user_permission{user_vhost = #user_vhost{username     = Username,
+                                                  virtual_host = VHostPath},
+                         permission = #permission{ configure = ConfigurePerm,
+                                                   write     = WritePerm,
+                                                   read      = ReadPerm}} <-
+            %% TODO: use dirty ops instead
+            rabbit_misc:execute_mnesia_transaction(QueryThunk)].
+
+match_user_vhost(Username, VHostPath) ->
+    fun () -> mnesia:match_object(
+                rabbit_user_permission,
+                #user_permission{user_vhost = #user_vhost{
+                                   username     = Username,
+                                   virtual_host = VHostPath},
+                                 permission = '_'},
+                read)
+    end.
diff --git a/src/rabbit_auth_mechanism_external.erl b/src/rabbit_auth_mechanism_external.erl
index 6572f786..1c4e5c15 100644
--- a/src/rabbit_auth_mechanism_external.erl
+++ b/src/rabbit_auth_mechanism_external.erl
@@ -81,7 +81,7 @@ handle_response(_Response, #state{username = Username}) ->
         {refused, _, _} = E ->
             E;
         _ ->
-            case rabbit_access_control:lookup_user(Username) of
+            case rabbit_access_control:check_user_login(Username, []) of
                 {ok, User} ->
                     {ok, User};
                 {error, not_found} ->
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 6f82d521..579d7f49 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -47,7 +47,7 @@
 -record(ch, {state, channel, reader_pid, writer_pid, limiter_pid,
              start_limiter_fun, transaction_id, tx_participants, next_tag,
              uncommitted_ack_q, unacked_message_q,
-             username, virtual_host, most_recently_declared_queue,
+             user, virtual_host, most_recently_declared_queue,
              consumer_mapping, blocking, queue_collector_pid, stats_timer,
              confirm_enabled, publish_seqno, unconfirmed, queues_for_msg}).
 
@@ -80,7 +80,7 @@
 -type(channel_number() :: non_neg_integer()).
 
 -spec(start_link/7 ::
-      (channel_number(), pid(), pid(), rabbit_access_control:username(),
+      (channel_number(), pid(), pid(), rabbit_types:user(),
        rabbit_types:vhost(), pid(),
        fun ((non_neg_integer()) -> rabbit_types:ok(pid()))) ->
                            rabbit_types:ok_pid_or_error()).
@@ -107,9 +107,9 @@
 
 %%----------------------------------------------------------------------------
 
-start_link(Channel, ReaderPid, WriterPid, Username, VHost, CollectorPid,
+start_link(Channel, ReaderPid, WriterPid, User, VHost, CollectorPid,
            StartLimiterFun) ->
-    gen_server2:start_link(?MODULE, [Channel, ReaderPid, WriterPid, Username,
+    gen_server2:start_link(?MODULE, [Channel, ReaderPid, WriterPid, User,
                                      VHost, CollectorPid, StartLimiterFun], []).
 
 do(Pid, Method) ->
@@ -161,7 +161,7 @@ emit_stats(Pid) ->
 
 %%---------------------------------------------------------------------------
 
-init([Channel, ReaderPid, WriterPid, Username, VHost, CollectorPid,
+init([Channel, ReaderPid, WriterPid, User, VHost, CollectorPid,
       StartLimiterFun]) ->
     process_flag(trap_exit, true),
     ok = pg_local:join(rabbit_channels, self()),
@@ -177,7 +177,7 @@ init([Channel, ReaderPid, WriterPid, Username, VHost, CollectorPid,
                 next_tag                = 1,
                 uncommitted_ack_q       = queue:new(),
                 unacked_message_q       = queue:new(),
-                username                = Username,
+                user                    = User,
                 virtual_host            = VHost,
                 most_recently_declared_queue = <<>>,
                 consumer_mapping        = dict:new(),
@@ -358,7 +358,7 @@ return_queue_declare_ok(#resource{name = ActualName},
                                   message_count  = MessageCount,
                                   consumer_count = ConsumerCount}).
 
-check_resource_access(Username, Resource, Perm) ->
+check_resource_access(User, Resource, Perm) ->
     V = {Resource, Perm},
     Cache = case get(permission_cache) of
                 undefined -> [];
@@ -368,7 +368,7 @@ check_resource_access(Username, Resource, Perm) ->
         case lists:member(V, Cache) of
             true  -> lists:delete(V, Cache);
             false -> ok = rabbit_access_control:check_resource_access(
-                            Username, Resource, Perm),
+                            User, Resource, Perm),
                      lists:sublist(Cache, ?MAX_PERMISSION_CACHE_SIZE - 1)
         end,
     put(permission_cache, [V | CacheTail]),
@@ -378,14 +378,25 @@ clear_permission_cache() ->
     erase(permission_cache),
     ok.
 
-check_configure_permitted(Resource, #ch{username = Username}) ->
-    check_resource_access(Username, Resource, configure).
+check_configure_permitted(Resource, #ch{user = User}) ->
+    check_resource_access(User, Resource, configure).
 
-check_write_permitted(Resource, #ch{username = Username}) ->
-    check_resource_access(Username, Resource, write).
+check_write_permitted(Resource, #ch{user = User}) ->
+    check_resource_access(User, Resource, write).
 
-check_read_permitted(Resource, #ch{username = Username}) ->
-    check_resource_access(Username, Resource, read).
+check_read_permitted(Resource, #ch{user = User}) ->
+    check_resource_access(User, Resource, read).
+
+check_user_id_header(#'P_basic'{user_id = undefined}, _) ->
+    ok;
+check_user_id_header(#'P_basic'{user_id = Username},
+                     #ch{user = #user{username = Username}}) ->
+    ok;
+check_user_id_header(#'P_basic'{user_id = Claimed},
+                     #ch{user = #user{username = Actual}}) ->
+    rabbit_misc:protocol_error(
+      precondition_failed, "user_id property set to '~s' but "
+      "authenticated user was '~s'", [Claimed, Actual]).
 
 check_internal_exchange(#exchange{name = Name, internal = true}) ->
     rabbit_misc:protocol_error(access_refused,
@@ -526,6 +537,7 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
     %% We decode the content's properties here because we're almost
     %% certain to want to look at delivery-mode and priority.
     DecodedContent = rabbit_binary_parser:ensure_content_decoded(Content),
+    check_user_id_header(DecodedContent#content.properties, State),
     IsPersistent = is_message_persistent(DecodedContent),
     {MsgSeqNo, State1}
         = case ConfirmEnabled of
@@ -1255,7 +1267,7 @@ infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items].
 i(pid,            _)                                 -> self();
 i(connection,     #ch{reader_pid       = ReaderPid}) -> ReaderPid;
 i(number,         #ch{channel          = Channel})   -> Channel;
-i(user,           #ch{username         = Username})  -> Username;
+i(user,           #ch{user             = User})      -> User#user.username;
 i(vhost,          #ch{virtual_host     = VHost})     -> VHost;
 i(transactional,  #ch{transaction_id   = TxnKey})    -> TxnKey =/= none;
 i(consumer_count, #ch{consumer_mapping = ConsumerMapping}) ->
diff --git a/src/rabbit_channel_sup.erl b/src/rabbit_channel_sup.erl
index 02199a65..a36253a0 100644
--- a/src/rabbit_channel_sup.erl
+++ b/src/rabbit_channel_sup.erl
@@ -48,7 +48,7 @@
 -type(start_link_args() ::
         {rabbit_types:protocol(), rabbit_net:socket(),
          rabbit_channel:channel_number(), non_neg_integer(), pid(),
-         rabbit_access_control:username(), rabbit_types:vhost(), pid()}).
+         rabbit_types:user(), rabbit_types:vhost(), pid()}).
 
 -spec(start_link/1 :: (start_link_args()) -> {'ok', pid(), pid()}).
 
@@ -56,7 +56,7 @@
 
 %%----------------------------------------------------------------------------
 
-start_link({Protocol, Sock, Channel, FrameMax, ReaderPid, Username, VHost,
+start_link({Protocol, Sock, Channel, FrameMax, ReaderPid, User, VHost,
             Collector}) ->
     {ok, SupPid} = supervisor2:start_link(?MODULE, []),
     {ok, WriterPid} =
@@ -69,7 +69,7 @@ start_link({Protocol, Sock, Channel, FrameMax, ReaderPid, Username, VHost,
         supervisor2:start_child(
           SupPid,
           {channel, {rabbit_channel, start_link,
-                     [Channel, ReaderPid, WriterPid, Username, VHost,
+                     [Channel, ReaderPid, WriterPid, User, VHost,
                       Collector, start_limiter_fun(SupPid)]},
            intrinsic, ?MAX_WAIT, worker, [rabbit_channel]}),
     {ok, FramingChannelPid} =
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index df55d961..8a3275bc 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -201,48 +201,48 @@ action(close_connection, Node, [PidStr, Explanation], _Opts, Inform) ->
 
 action(add_user, Node, Args = [Username, _Password], _Opts, Inform) ->
     Inform("Creating user ~p", [Username]),
-    call(Node, {rabbit_access_control, add_user, Args});
+    call(Node, {rabbit_auth_backend_internal, add_user, Args});
 
 action(delete_user, Node, Args = [_Username], _Opts, Inform) ->
     Inform("Deleting user ~p", Args),
-    call(Node, {rabbit_access_control, delete_user, Args});
+    call(Node, {rabbit_auth_backend_internal, delete_user, Args});
 
 action(change_password, Node, Args = [Username, _Newpassword], _Opts, Inform) ->
     Inform("Changing password for user ~p", [Username]),
-    call(Node, {rabbit_access_control, change_password, Args});
+    call(Node, {rabbit_auth_backend_internal, change_password, Args});
 
 action(clear_password, Node, Args = [Username], _Opts, Inform) ->
     Inform("Clearing password for user ~p", [Username]),
-    call(Node, {rabbit_access_control, clear_password, Args});
+    call(Node, {rabbit_auth_backend_internal, clear_password, Args});
 
 action(set_admin, Node, [Username], _Opts, Inform) ->
     Inform("Setting administrative status for user ~p", [Username]),
-    call(Node, {rabbit_access_control, set_admin, [Username]});
+    call(Node, {rabbit_auth_backend_internal, set_admin, [Username]});
 
 action(clear_admin, Node, [Username], _Opts, Inform) ->
     Inform("Clearing administrative status for user ~p", [Username]),
-    call(Node, {rabbit_access_control, clear_admin, [Username]});
+    call(Node, {rabbit_auth_backend_internal, clear_admin, [Username]});
 
 action(list_users, Node, [], _Opts, Inform) ->
     Inform("Listing users", []),
-    display_list(call(Node, {rabbit_access_control, list_users, []}));
+    display_list(call(Node, {rabbit_auth_backend_internal, list_users, []}));
 
 action(add_vhost, Node, Args = [_VHostPath], _Opts, Inform) ->
     Inform("Creating vhost ~p", Args),
-    call(Node, {rabbit_access_control, add_vhost, Args});
+    call(Node, {rabbit_vhost, add, Args});
 
 action(delete_vhost, Node, Args = [_VHostPath], _Opts, Inform) ->
     Inform("Deleting vhost ~p", Args),
-    call(Node, {rabbit_access_control, delete_vhost, Args});
+    call(Node, {rabbit_vhost, delete, Args});
 
 action(list_vhosts, Node, [], _Opts, Inform) ->
     Inform("Listing vhosts", []),
-    display_list(call(Node, {rabbit_access_control, list_vhosts, []}));
+    display_list(call(Node, {rabbit_vhost, list, []}));
 
 action(list_user_permissions, Node, Args = [_Username], _Opts, Inform) ->
     Inform("Listing permissions for user ~p", Args),
-    display_list(call(Node, {rabbit_access_control, list_user_permissions,
-                             Args}));
+    display_list(call(Node, {rabbit_auth_backend_internal,
+                             list_user_permissions, Args}));
 
 action(list_queues, Node, Args, Opts, Inform) ->
     Inform("Listing queues", []),
@@ -296,19 +296,20 @@ action(list_consumers, Node, _Args, Opts, Inform) ->
 action(set_permissions, Node, [Username, CPerm, WPerm, RPerm], Opts, Inform) ->
     VHost = proplists:get_value(?VHOST_OPT, Opts),
     Inform("Setting permissions for user ~p in vhost ~p", [Username, VHost]),
-    call(Node, {rabbit_access_control, set_permissions,
+    call(Node, {rabbit_auth_backend_internal, set_permissions,
                 [Username, VHost, CPerm, WPerm, RPerm]});
 
 action(clear_permissions, Node, [Username], Opts, Inform) ->
     VHost = proplists:get_value(?VHOST_OPT, Opts),
     Inform("Clearing permissions for user ~p in vhost ~p", [Username, VHost]),
-    call(Node, {rabbit_access_control, clear_permissions, [Username, VHost]});
+    call(Node, {rabbit_auth_backend_internal, clear_permissions,
+                [Username, VHost]});
 
 action(list_permissions, Node, [], Opts, Inform) ->
     VHost = proplists:get_value(?VHOST_OPT, Opts),
     Inform("Listing permissions in vhost ~p", [VHost]),
-    display_list(call(Node, {rabbit_access_control, list_vhost_permissions,
-                             [VHost]})).
+    display_list(call(Node, {rabbit_auth_backend_internal,
+                             list_vhost_permissions, [VHost]})).
 
 default_if_empty(List, Default) when is_list(List) ->
     if List == [] ->
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 06ba319b..15ba787a 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -46,7 +46,7 @@
 -export([enable_cover/1, report_cover/1]).
 -export([start_cover/1]).
 -export([throw_on_error/2, with_exit_handler/2, filter_exit_map/2]).
--export([with_user/2, with_vhost/2, with_user_and_vhost/3]).
+-export([with_user/2, with_user_and_vhost/3]).
 -export([execute_mnesia_transaction/1]).
 -export([ensure_ok/2]).
 -export([makenode/1, nodeparts/1, cookie_hash/0, tcp_name/3]).
@@ -72,7 +72,7 @@
 
 -ifdef(use_specs).
 
--export_type([resource_name/0]).
+-export_type([resource_name/0, thunk/1]).
 
 -type(ok_or_error() :: rabbit_types:ok_or_error(any())).
 -type(thunk(T) :: fun(() -> T)).
@@ -137,10 +137,9 @@
         (atom(), thunk(rabbit_types:error(any()) | {ok, A} | A)) -> A).
 -spec(with_exit_handler/2 :: (thunk(A), thunk(A)) -> A).
 -spec(filter_exit_map/2 :: (fun ((A) -> B), [A]) -> [B]).
--spec(with_user/2 :: (rabbit_access_control:username(), thunk(A)) -> A).
--spec(with_vhost/2 :: (rabbit_types:vhost(), thunk(A)) -> A).
+-spec(with_user/2 :: (rabbit_types:username(), thunk(A)) -> A).
 -spec(with_user_and_vhost/3 ::
-        (rabbit_access_control:username(), rabbit_types:vhost(), thunk(A))
+        (rabbit_types:username(), rabbit_types:vhost(), thunk(A))
         -> A).
 -spec(execute_mnesia_transaction/1 :: (thunk(A)) -> A).
 -spec(ensure_ok/2 :: (ok_or_error(), atom()) -> 'ok').
@@ -366,19 +365,8 @@ with_user(Username, Thunk) ->
             end
     end.
 
-with_vhost(VHostPath, Thunk) ->
-    fun () ->
-            case mnesia:read({rabbit_vhost, VHostPath}) of
-                [] ->
-                    mnesia:abort({no_such_vhost, VHostPath});
-                [_V] ->
-                    Thunk()
-            end
-    end.
-
 with_user_and_vhost(Username, VHostPath, Thunk) ->
-    with_user(Username, with_vhost(VHostPath, Thunk)).
-
+    with_user(Username, rabbit_vhost:with(VHostPath, Thunk)).
 
 execute_mnesia_transaction(TxFun) ->
     %% Making this a sync_transaction allows us to use dirty_read
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 11f5e410..38cc82a6 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -163,10 +163,10 @@ nodes_of_type(Type) ->
 
 table_definitions() ->
     [{rabbit_user,
-      [{record_name, user},
-       {attributes, record_info(fields, user)},
+      [{record_name, internal_user},
+       {attributes, record_info(fields, internal_user)},
        {disc_copies, [node()]},
-       {match, #user{_='_'}}]},
+       {match, #internal_user{_='_'}}]},
      {rabbit_user_permission,
       [{record_name, user_permission},
        {attributes, record_info(fields, user_permission)},
diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl
index c50be9c9..b20b61e7 100644
--- a/src/rabbit_msg_store.erl
+++ b/src/rabbit_msg_store.erl
@@ -727,7 +727,8 @@ handle_cast({write, CRef, Guid},
             ok = index_update_ref_count(Guid, RefCount + 1, State1),
             CTG2 = case {dict:find(CRef, CODC), File} of
                        {{ok, _},   CurFile} -> CTG1;
-                       {{ok, Fun}, _}       -> Fun(gb_sets:singleton(Guid)),
+                       {{ok, Fun}, _}       -> Fun(gb_sets:singleton(Guid),
+                                                   written),
                                                CTG;
                        _                    -> CTG1
                    end,
@@ -735,12 +736,10 @@ handle_cast({write, CRef, Guid},
     end;
 
 handle_cast({remove, CRef, Guids}, State) ->
-    State1 = lists:foldl(
-               fun (Guid, State2) -> remove_message(Guid, State2) end,
-               State, Guids),
-    State2 = client_confirm(CRef, gb_sets:from_list(Guids),
-                            false, State1),
-    noreply(maybe_compact(State2));
+    State1 = lists:foldl(fun (Guid, State2) -> remove_message(Guid, State2) end,
+                         State, Guids),
+    State3 = client_confirm(CRef, gb_sets:from_list(Guids), removed, State1),
+    noreply(maybe_compact(State3));
 
 handle_cast({release, Guids}, State =
                 #msstate { dedup_cache_ets = DedupCacheEts }) ->
@@ -876,7 +875,7 @@ internal_sync(State = #msstate { current_file_handle    = CurHdl,
        true                            -> file_handle_cache:sync(CurHdl)
     end,
     lists:foreach(fun (K) -> K() end, lists:reverse(Syncs)),
-    [client_confirm(CRef, Guids, true, State1)
+    [client_confirm(CRef, Guids, written, State1)
      || {CRef, Guids} <- CGs],
     State1 #msstate { cref_to_guids = dict:new(), on_sync = [] }.
 
@@ -1057,11 +1056,11 @@ orddict_store(Key, Val, Dict) ->
     false = orddict:is_key(Key, Dict),
     orddict:store(Key, Val, Dict).
 
-client_confirm(CRef, Guids, WaitForIndex,
+client_confirm(CRef, Guids, ActionTaken,
                State = #msstate { client_ondisk_callback = CODC,
                                   cref_to_guids          = CTG }) ->
     case dict:find(CRef, CODC) of
-        {ok, Fun} -> Fun(Guids, WaitForIndex),
+        {ok, Fun} -> Fun(Guids, ActionTaken),
                      CTG1 = case dict:find(CRef, CTG) of
                                 {ok, Gs} ->
                                     Guids1 = gb_sets:difference(Gs, Guids),
diff --git a/src/rabbit_net.erl b/src/rabbit_net.erl
index 89954b06..c6a083bb 100644
--- a/src/rabbit_net.erl
+++ b/src/rabbit_net.erl
@@ -32,7 +32,7 @@
 -module(rabbit_net).
 -include("rabbit.hrl").
 
--export([is_ssl/1, controlling_process/2, getstat/2,
+-export([is_ssl/1, ssl_info/1, controlling_process/2, getstat/2,
          async_recv/3, port_command/2, send/2, close/1,
          sockname/1, peername/1, peercert/1]).
 
@@ -50,6 +50,9 @@
 -type(socket() :: port() | #ssl_socket{}).
 
 -spec(is_ssl/1 :: (socket()) -> boolean()).
+-spec(ssl_info/1 :: (socket())
+                    -> 'nossl' | ok_val_or_error(
+                                   {atom(), {atom(), atom(), atom()}})).
 -spec(controlling_process/2 :: (socket(), pid()) -> ok_or_any_error()).
 -spec(getstat/2 ::
         (socket(), [stat_option()])
@@ -77,6 +80,11 @@
 
 is_ssl(Sock) -> ?IS_SSL(Sock).
 
+ssl_info(Sock) when ?IS_SSL(Sock) ->
+    ssl:connection_info(Sock#ssl_socket.ssl);
+ssl_info(_Sock) ->
+    nossl.
+
 controlling_process(Sock, Pid) when ?IS_SSL(Sock) ->
     ssl:controlling_process(Sock#ssl_socket.ssl, Pid);
 controlling_process(Sock, Pid) when is_port(Sock) ->
diff --git a/src/rabbit_queue_index.erl b/src/rabbit_queue_index.erl
index 76c0a4ef..2162104f 100644
--- a/src/rabbit_queue_index.erl
+++ b/src/rabbit_queue_index.erl
@@ -33,7 +33,7 @@
 
 -export([init/2, shutdown_terms/1, recover/5,
          terminate/2, delete_and_terminate/1,
-         publish/5, deliver/2, ack/2, sync/2, flush/1, read/3,
+         publish/5, deliver/2, ack/2, sync/1, sync/2, flush/1, read/3,
          next_segment_boundary/1, bounds/1, recover/1]).
 
 -export([add_queue_ttl/0]).
@@ -297,11 +297,12 @@ deliver(SeqIds, State) ->
 ack(SeqIds, State) ->
     deliver_or_ack(ack, SeqIds, State).
 
-sync([], State) ->
-    State;
-sync(_SeqIds, State = #qistate { journal_handle = undefined }) ->
-    State;
-sync(_SeqIds, State = #qistate { journal_handle = JournalHdl }) ->
+%% This is only called when there are outstanding confirms and the
+%% queue is idle.
+sync(State = #qistate { unsynced_guids = Guids }) ->
+    sync_if([] =/= Guids, State).
+
+sync(SeqIds, State) ->
     %% The SeqIds here contains the SeqId of every publish and ack in
     %% the transaction. Ideally we should go through these seqids and
     %% only sync the journal if the pubs or acks appear in the
@@ -309,9 +310,8 @@ sync(_SeqIds, State = #qistate { journal_handle = JournalHdl }) ->
     %% the variable queue publishes and acks to the qi, and then
     %% syncs, all in one operation, there is no possibility of the
     %% seqids not being in the journal, provided the transaction isn't
-    %% emptied (handled above anyway).
-    ok = file_handle_cache:sync(JournalHdl),
-    notify_sync(State).
+    %% emptied (handled by sync_if anyway).
+    sync_if([] =/= SeqIds, State).
 
 flush(State = #qistate { dirty_count = 0 }) -> State;
 flush(State)                                -> flush_journal(State).
@@ -723,6 +723,14 @@ deliver_or_ack(Kind, SeqIds, State) ->
                                             add_to_journal(SeqId, Kind, StateN)
                                     end, State1, SeqIds)).
 
+sync_if(false, State) ->
+    State;
+sync_if(_Bool, State = #qistate { journal_handle = undefined }) ->
+    State;
+sync_if(true, State = #qistate { journal_handle = JournalHdl }) ->
+    ok = file_handle_cache:sync(JournalHdl),
+    notify_sync(State).
+
 notify_sync(State = #qistate { unsynced_guids = UG, on_sync = OnSyncFun }) ->
     OnSyncFun(gb_sets:from_list(UG)),
     State #qistate { unsynced_guids = [] }.
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 92a2f4d7..a3292d35 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -65,6 +65,8 @@
 -define(CREATION_EVENT_KEYS, [pid, address, port, peer_address, peer_port, ssl,
                               peer_cert_subject, peer_cert_issuer,
                               peer_cert_validity, auth_mechanism,
+                              ssl_protocol, ssl_key_exchange,
+                              ssl_cipher, ssl_hash,
                               protocol, user, vhost, timeout, frame_max,
                               client_properties]).
 
@@ -905,6 +907,14 @@ i(peer_port, #v1{sock = Sock}) ->
     socket_info(fun rabbit_net:peername/1, fun ({_, P}) -> P end, Sock);
 i(ssl, #v1{sock = Sock}) ->
     rabbit_net:is_ssl(Sock);
+i(ssl_protocol, #v1{sock = Sock}) ->
+    ssl_info(fun ({P, _}) -> P end, Sock);
+i(ssl_key_exchange, #v1{sock = Sock}) ->
+    ssl_info(fun ({_, {K, _, _}}) -> K end, Sock);
+i(ssl_cipher, #v1{sock = Sock}) ->
+    ssl_info(fun ({_, {_, C, _}}) -> C end, Sock);
+i(ssl_hash, #v1{sock = Sock}) ->
+    ssl_info(fun ({_, {_, _, H}}) -> H end, Sock);
 i(peer_cert_issuer, #v1{sock = Sock}) ->
     cert_info(fun rabbit_ssl:peer_cert_issuer/1, Sock);
 i(peer_cert_subject, #v1{sock = Sock}) ->
@@ -955,6 +965,13 @@ socket_info(Get, Select) ->
         {error, _} -> ''
     end.
 
+ssl_info(F, Sock) ->
+    case rabbit_net:ssl_info(Sock) of
+        nossl       -> '';
+        {error, _}  -> '';
+        {ok, Info}  -> F(Info)
+    end.
+
 cert_info(F, Sock) ->
     case rabbit_net:peercert(Sock) of
         nossl                -> '';
@@ -969,12 +986,12 @@ send_to_new_channel(Channel, AnalyzedFrame, State) ->
         channel_sup_sup_pid = ChanSupSup,
         connection = #connection{protocol  = Protocol,
                                  frame_max = FrameMax,
-                                 user      = #user{username = Username},
+                                 user      = User,
                                  vhost     = VHost}} = State,
     {ok, ChSupPid, ChFrPid} =
         rabbit_channel_sup_sup:start_channel(
           ChanSupSup, {Protocol, Sock, Channel, FrameMax,
-                       self(), Username, VHost, Collector}),
+                       self(), User, VHost, Collector}),
     erlang:monitor(process, ChSupPid),
     put({channel, Channel}, {ch_fr_pid, ChFrPid}),
     put({ch_sup_pid, ChSupPid}, {{channel, Channel}, {ch_fr_pid, ChFrPid}}),
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 1f320a10..9614dc90 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -1030,7 +1030,7 @@ test_server_status() ->
     %% create a few things so there is some useful information to list
     Writer = spawn(fun () -> receive shutdown -> ok end end),
     {ok, Ch} = rabbit_channel:start_link(1, self(), Writer,
-                                         <<"user">>, <<"/">>, self(),
+                                         user(<<"user">>), <<"/">>, self(),
                                          fun (_) -> {ok, self()} end),
     [Q, Q2] = [Queue || Name <- [<<"foo">>, <<"bar">>],
                         {new, Queue = #amqqueue{}} <-
@@ -1090,7 +1090,7 @@ test_spawn(Receiver) ->
     Me = self(),
     Writer = spawn(fun () -> Receiver(Me) end),
     {ok, Ch} = rabbit_channel:start_link(1, Me, Writer,
-                                         <<"guest">>, <<"/">>, self(),
+                                         user(<<"guest">>), <<"/">>, self(),
                                          fun (_) -> {ok, self()} end),
     ok = rabbit_channel:do(Ch, #'channel.open'{}),
     receive #'channel.open_ok'{} -> ok
@@ -1098,6 +1098,13 @@ test_spawn(Receiver) ->
     end,
     {Writer, Ch}.
 
+user(Username) ->
+    #user{username     = Username,
+          is_admin     = true,
+          auth_backend = rabbit_auth_backend_internal,
+          impl         = #internal_user{username = Username,
+                                        is_admin = true}}.
+
 test_statistics_receiver(Pid) ->
     receive
         shutdown ->
diff --git a/src/rabbit_types.erl b/src/rabbit_types.erl
index 548014be..70d18d7a 100644
--- a/src/rabbit_types.erl
+++ b/src/rabbit_types.erl
@@ -42,8 +42,9 @@
               vhost/0, ctag/0, amqp_error/0, r/1, r2/2, r3/3, listener/0,
               binding/0, binding_source/0, binding_destination/0,
               amqqueue/0, exchange/0,
-              connection/0, protocol/0, user/0, ok/1, error/1, ok_or_error/1,
-              ok_or_error2/2, ok_pid_or_error/0, channel_exit/0,
+              connection/0, protocol/0, user/0, internal_user/0,
+              username/0, password/0, password_hash/0, ok/1, error/1,
+              ok_or_error/1, ok_or_error2/2, ok_pid_or_error/0, channel_exit/0,
               connection_exit/0]).
 
 -type(channel_exit() :: no_return()).
@@ -151,9 +152,19 @@
 -type(protocol() :: rabbit_framing:protocol()).
 
 -type(user() ::
-      #user{username      :: rabbit_access_control:username(),
-            password_hash :: rabbit_access_control:password_hash(),
-            is_admin      :: boolean()}).
+        #user{username     :: username(),
+              is_admin     :: boolean(),
+              auth_backend :: atom(),
+              impl         :: any()}).
+
+-type(internal_user() ::
+        #internal_user{username      :: username(),
+                       password_hash :: password_hash(),
+                       is_admin      :: boolean()}).
+
+-type(username() :: binary()).
+-type(password() :: binary()).
+-type(password_hash() :: binary()).
 
 -type(ok(A) :: {'ok', A}).
 -type(error(A) :: {'error', A}).
diff --git a/src/rabbit_upgrade_functions.erl b/src/rabbit_upgrade_functions.erl
index 7848c848..b5ff2b12 100644
--- a/src/rabbit_upgrade_functions.erl
+++ b/src/rabbit_upgrade_functions.erl
@@ -28,6 +28,7 @@
 -rabbit_upgrade({hash_passwords,     []}).
 -rabbit_upgrade({add_ip_to_listener, []}).
 -rabbit_upgrade({internal_exchanges, []}).
+-rabbit_upgrade({user_to_internal_user, [hash_passwords]}).
 
 %% -------------------------------------------------------------------
 
@@ -37,6 +38,7 @@
 -spec(hash_passwords/0     :: () -> 'ok').
 -spec(add_ip_to_listener/0 :: () -> 'ok').
 -spec(internal_exchanges/0 :: () -> 'ok').
+-spec(user_to_internal_user/0 :: () -> 'ok').
 
 -endif.
 
@@ -60,7 +62,7 @@ hash_passwords() ->
     mnesia(
       rabbit_user,
       fun ({user, Username, Password, IsAdmin}) ->
-              Hash = rabbit_access_control:hash_password(Password),
+              Hash = rabbit_auth_backend_internal:hash_password(Password),
               {user, Username, Hash, IsAdmin}
       end,
       [username, password_hash, is_admin]).
@@ -85,8 +87,21 @@ internal_exchanges() ->
       || T <- Tables ],
     ok.
 
+user_to_internal_user() ->
+    mnesia(
+      rabbit_user,
+      fun({user, Username, PasswordHash, IsAdmin}) ->
+              {internal_user, Username, PasswordHash, IsAdmin}
+      end,
+      [username, password_hash, is_admin], internal_user).
+
 %%--------------------------------------------------------------------
 
 mnesia(TableName, Fun, FieldList) ->
     {atomic, ok} = mnesia:transform_table(TableName, Fun, FieldList),
     ok.
+
+mnesia(TableName, Fun, FieldList, NewRecordName) ->
+    {atomic, ok} = mnesia:transform_table(TableName, Fun, FieldList,
+                                          NewRecordName),
+    ok.
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index c30be37c..18423dd7 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -412,8 +412,8 @@ stop_msg_store() ->
 init(QueueName, IsDurable, Recover) ->
     Self = self(),
     init(QueueName, IsDurable, Recover,
-         fun (Guids, WaitForIndex) ->
-                 msgs_written_to_disk(Self, Guids, WaitForIndex)
+         fun (Guids, ActionTaken) ->
+                 msgs_written_to_disk(Self, Guids, ActionTaken)
          end,
          fun (Guids) -> msg_indices_written_to_disk(Self, Guids) end).
 
@@ -535,20 +535,20 @@ publish_delivered(true, Msg = #basic_message { is_persistent = IsPersistent,
                                      in_counter       = InCount,
                                      persistent_count = PCount,
                                      durable          = IsDurable,
-                                     unconfirmed      = Unconfirmed }) ->
+                                     unconfirmed      = UC }) ->
     IsPersistent1 = IsDurable andalso IsPersistent,
     MsgStatus = (msg_status(IsPersistent1, SeqId, Msg, MsgProps))
         #msg_status { is_delivered = true },
     {MsgStatus1, State1} = maybe_write_to_disk(false, false, MsgStatus, State),
     State2 = record_pending_ack(m(MsgStatus1), State1),
     PCount1 = PCount + one_if(IsPersistent1),
-    Unconfirmed1 = gb_sets_maybe_insert(NeedsConfirming, Guid, Unconfirmed),
+    UC1 = gb_sets_maybe_insert(NeedsConfirming, Guid, UC),
     {SeqId, a(reduce_memory_use(
                 State2 #vqstate { next_seq_id      = SeqId    + 1,
                                   out_counter      = OutCount + 1,
                                   in_counter       = InCount  + 1,
                                   persistent_count = PCount1,
-                                  unconfirmed      = Unconfirmed1 }))}.
+                                  unconfirmed      = UC1 }))}.
 
 dropwhile(Pred, State) ->
     {_OkOrEmpty, State1} = dropwhile1(Pred, State),
@@ -809,17 +809,22 @@ ram_duration(State = #vqstate {
                  ram_msg_count_prev = RamMsgCount,
                  ram_ack_count_prev = RamAckCount }}.
 
-needs_idle_timeout(State = #vqstate { on_sync = ?BLANK_SYNC }) ->
-    {Res, _State} = reduce_memory_use(fun (_Quota, State1) -> {0, State1} end,
-                                      fun (_Quota, State1) -> State1 end,
-                                      fun (State1)         -> State1 end,
-                                      fun (_Quota, State1) -> {0, State1} end,
-                                      State),
-    Res;
-needs_idle_timeout(_State) ->
-    true.
+needs_idle_timeout(State = #vqstate { on_sync = OnSync, unconfirmed = UC }) ->
+    case {OnSync, gb_sets:is_empty(UC)} of
+        {?BLANK_SYNC, true} ->
+            {Res, _State} = reduce_memory_use(
+                              fun (_Quota, State1) -> {0, State1} end,
+                              fun (_Quota, State1) -> State1 end,
+                              fun (State1)         -> State1 end,
+                              fun (_Quota, State1) -> {0, State1} end,
+                              State),
+            Res;
+        _ ->
+            true
+    end.
 
-idle_timeout(State) -> a(reduce_memory_use(tx_commit_index(State))).
+idle_timeout(State) ->
+    a(reduce_memory_use(confirm_commit_index(tx_commit_index(State)))).
 
 handle_pre_hibernate(State = #vqstate { index_state = IndexState }) ->
     State #vqstate { index_state = rabbit_queue_index:flush(IndexState) }.
@@ -1232,7 +1237,7 @@ publish(Msg = #basic_message { is_persistent = IsPersistent, guid = Guid },
                            persistent_count = PCount,
                            durable          = IsDurable,
                            ram_msg_count    = RamMsgCount,
-                           unconfirmed      = Unconfirmed }) ->
+                           unconfirmed      = UC }) ->
     IsPersistent1 = IsDurable andalso IsPersistent,
     MsgStatus = (msg_status(IsPersistent1, SeqId, Msg, MsgProps))
         #msg_status { is_delivered = IsDelivered, msg_on_disk = MsgOnDisk},
@@ -1242,13 +1247,13 @@ publish(Msg = #basic_message { is_persistent = IsPersistent, guid = Guid },
                  true  -> State1 #vqstate { q4 = queue:in(m(MsgStatus1), Q4) }
              end,
     PCount1 = PCount + one_if(IsPersistent1),
-    Unconfirmed1 = gb_sets_maybe_insert(NeedsConfirming, Guid, Unconfirmed),
+    UC1 = gb_sets_maybe_insert(NeedsConfirming, Guid, UC),
     {SeqId, State2 #vqstate { next_seq_id      = SeqId   + 1,
                               len              = Len     + 1,
                               in_counter       = InCount + 1,
                               persistent_count = PCount1,
                               ram_msg_count    = RamMsgCount + 1,
-                              unconfirmed      = Unconfirmed1 }}.
+                              unconfirmed      = UC1 }}.
 
 maybe_write_msg_to_disk(_Force, MsgStatus = #msg_status {
                                   msg_on_disk = true }, _MSCState) ->
@@ -1386,6 +1391,11 @@ find_persistent_count(LensByStore) ->
 %% Internal plumbing for confirms (aka publisher acks)
 %%----------------------------------------------------------------------------
 
+confirm_commit_index(State = #vqstate { unconfirmed = [] }) ->
+    State;
+confirm_commit_index(State = #vqstate { index_state = IndexState }) ->
+    State #vqstate { index_state = rabbit_queue_index:sync(IndexState) }.
+
 remove_confirms(GuidSet, State = #vqstate { msgs_on_disk        = MOD,
                                             msg_indices_on_disk = MIOD,
                                             unconfirmed         = UC }) ->
@@ -1398,13 +1408,11 @@ msgs_confirmed(GuidSet, State) ->
 
 blind_confirm(QPid, GuidSet) ->
     rabbit_amqqueue:maybe_run_queue_via_backing_queue_async(
-      QPid, fun (State) ->
-                    msgs_confirmed(GuidSet, State)
-            end).
+      QPid, fun (State) -> msgs_confirmed(GuidSet, State) end).
 
-msgs_written_to_disk(QPid, GuidSet, false) ->
+msgs_written_to_disk(QPid, GuidSet, removed) ->
     blind_confirm(QPid, GuidSet);
-msgs_written_to_disk(QPid, GuidSet, true) ->
+msgs_written_to_disk(QPid, GuidSet, written) ->
     rabbit_amqqueue:maybe_run_queue_via_backing_queue_async(
       QPid, fun (State = #vqstate { msgs_on_disk        = MOD,
                                     msg_indices_on_disk = MIOD,
diff --git a/src/rabbit_vhost.erl b/src/rabbit_vhost.erl
new file mode 100644
index 00000000..f939a3fe
--- /dev/null
+++ b/src/rabbit_vhost.erl
@@ -0,0 +1,122 @@
+%%   The contents of this file are subject to the Mozilla Public License
+%%   Version 1.1 (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.mozilla.org/MPL/
+%%
+%%   Software distributed under the License is distributed on an "AS IS"
+%%   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+%%   License for the specific language governing rights and limitations
+%%   under the License.
+%%
+%%   The Original Code is RabbitMQ.
+%%
+%%   The Initial Developers of the Original Code are LShift Ltd,
+%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+%%   Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
+%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
+%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%%   (C) 2007-2010 Rabbit Technologies Ltd.
+%%
+%%   All Rights Reserved.
+%%
+%%   Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_vhost).
+
+-include("rabbit.hrl").
+
+%%----------------------------------------------------------------------------
+
+-export([add/1, delete/1, exists/1, list/0, with/2]).
+
+-ifdef(use_specs).
+
+-spec(add/1 :: (rabbit_types:vhost()) -> 'ok').
+-spec(delete/1 :: (rabbit_types:vhost()) -> 'ok').
+-spec(exists/1 :: (rabbit_types:vhost()) -> boolean()).
+-spec(list/0 :: () -> [rabbit_types:vhost()]).
+-spec(with/2 :: (rabbit_types:vhost(), rabbit_misc:thunk(A)) -> A).
+
+-endif.
+
+%%----------------------------------------------------------------------------
+
+add(VHostPath) ->
+    R = rabbit_misc:execute_mnesia_transaction(
+          fun () ->
+                  case mnesia:wread({rabbit_vhost, VHostPath}) of
+                      [] ->
+                          ok = mnesia:write(rabbit_vhost,
+                                            #vhost{virtual_host = VHostPath},
+                                            write),
+                          [rabbit_exchange:declare(
+                             rabbit_misc:r(VHostPath, exchange, Name),
+                             Type, true, false, false, []) ||
+                              {Name,Type} <-
+                                  [{<<"">>,           direct},
+                                   {<<"amq.direct">>, direct},
+                                   {<<"amq.topic">>,  topic},
+                                   {<<"amq.match">>,  headers}, %% per 0-9-1 pdf
+                                   {<<"amq.headers">>,  headers}, %% per 0-9-1 xml
+                                   {<<"amq.fanout">>, fanout}]],
+                          ok;
+                      [_] ->
+                          mnesia:abort({vhost_already_exists, VHostPath})
+                  end
+          end),
+    rabbit_log:info("Added vhost ~p~n", [VHostPath]),
+    R.
+
+delete(VHostPath) ->
+    %%FIXME: We are forced to delete the queues outside the TX below
+    %%because queue deletion involves sending messages to the queue
+    %%process, which in turn results in further mnesia actions and
+    %%eventually the termination of that process.
+    lists:foreach(fun (Q) ->
+                          {ok,_} = rabbit_amqqueue:delete(Q, false, false)
+                  end,
+                  rabbit_amqqueue:list(VHostPath)),
+    R = rabbit_misc:execute_mnesia_transaction(
+          with(VHostPath, fun () ->
+                                  ok = internal_delete(VHostPath)
+                          end)),
+    rabbit_log:info("Deleted vhost ~p~n", [VHostPath]),
+    R.
+
+internal_delete(VHostPath) ->
+    lists:foreach(fun (#exchange{name = Name}) ->
+                          ok = rabbit_exchange:delete(Name, false)
+                  end,
+                  rabbit_exchange:list(VHostPath)),
+    lists:foreach(
+      fun ({Username, _, _, _}) ->
+              ok = rabbit_auth_backend_internal:clear_permissions(Username,
+                                                                  VHostPath)
+      end,
+      rabbit_auth_backend_internal:list_vhost_permissions(VHostPath)),
+    ok = mnesia:delete({rabbit_vhost, VHostPath}),
+    ok.
+
+exists(VHostPath) ->
+    mnesia:dirty_read({rabbit_vhost, VHostPath}) /= [].
+
+list() ->
+    mnesia:dirty_all_keys(rabbit_vhost).
+
+with(VHostPath, Thunk) ->
+    fun () ->
+            case mnesia:read({rabbit_vhost, VHostPath}) of
+                [] ->
+                    mnesia:abort({no_such_vhost, VHostPath});
+                [_V] ->
+                    Thunk()
+            end
+    end.