Extract rabbit_definitions from rabbitmq-management

Part of rabbitmq/rabbitmq-management#749.

4 files changed, 625 insertions, 1 deletions

diff --git a/src/rabbit.erl b/src/rabbit.erl
index 87ceb5d701..e4d2ee9808 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -187,6 +187,15 @@
                     {requires,    [core_initialized]},
                     {enables,     routing_ready}]}).
 
+%% We want to A) make sure we apply definitions before the node begins serving
+%% traffic and B) in fact do it before empty_db_check (so the defaults will not
+%% get created if we don't need 'em).
+-rabbit_boot_step({load_core_definitions,
+                   [{description, "imports definitions"},
+                    {mfa,         {rabbit_definitions, maybe_load_definitions, []}},
+                    {requires,    recovery},
+                    {enables,     empty_db_check}]}).
+
 -rabbit_boot_step({empty_db_check,
                    [{description, "empty DB check"},
                     {mfa,         {?MODULE, maybe_insert_default_data, []}},
diff --git a/src/rabbit_auth_backend_internal.erl b/src/rabbit_auth_backend_internal.erl
index e675ad188b..8777774775 100644
--- a/src/rabbit_auth_backend_internal.erl
+++ b/src/rabbit_auth_backend_internal.erl
@@ -28,7 +28,7 @@
          hash_password/2, change_password_hash/2, change_password_hash/3,
          set_tags/3, set_permissions/6, clear_permissions/3,
          set_topic_permissions/6, clear_topic_permissions/3, clear_topic_permissions/4,
-         add_user_sans_validation/3]).
+         add_user_sans_validation/3, put_user/2, put_user/3]).
 
 -export([user_info_keys/0, perms_info_keys/0,
          user_perms_info_keys/0, vhost_perms_info_keys/0,
@@ -471,6 +471,138 @@ clear_topic_permissions(Username, VHostPath, Exchange, ActingUser) ->
         {user_who_performed_action, ActingUser}]),
     R.
 
+put_user(User, ActingUser) -> put_user(User, undefined, ActingUser).
+
+put_user(User, Version, ActingUser) ->
+    Username        = maps:get(name, User),
+    HasPassword     = maps:is_key(password, User),
+    HasPasswordHash = maps:is_key(password_hash, User),
+    Password        = maps:get(password, User, undefined),
+    PasswordHash    = maps:get(password_hash, User, undefined),
+
+    Tags            = case {maps:get(tags, User, undefined), maps:get(administrator, User, undefined)} of
+                          {undefined, undefined} ->
+                              throw({error, tags_not_present});
+                          {undefined, AdminS} ->
+                              case rabbit_mgmt_util:parse_bool(AdminS) of
+                                  true  -> [administrator];
+                                  false -> []
+                              end;
+                          {TagsS, _} ->
+                              [list_to_atom(string:strip(T)) ||
+                                  T <- string:tokens(binary_to_list(TagsS), ",")]
+                      end,
+
+    UserExists      = case rabbit_auth_backend_internal:lookup_user(Username) of
+                          %% expected
+                          {error, not_found} -> false;
+                          %% shouldn't normally happen but worth guarding
+                          %% against
+                          {error, _}         -> false;
+                          _                  -> true
+                      end,
+
+    %% pre-configured, only applies to newly created users
+    Permissions     = maps:get(permissions, User, undefined),
+
+    PassedCredentialValidation =
+        case {HasPassword, HasPasswordHash} of
+            {true, false} ->
+                rabbit_credential_validation:validate(Username, Password) =:= ok;
+            {false, true} -> true;
+            _             ->
+                rabbit_credential_validation:validate(Username, Password) =:= ok
+        end,
+
+    case UserExists of
+        true  ->
+            case {HasPassword, HasPasswordHash} of
+                {true, false} ->
+                    update_user_password(PassedCredentialValidation, Username, Password, Tags, ActingUser);
+                {false, true} ->
+                    update_user_password_hash(Username, PasswordHash, Tags, User, Version, ActingUser);
+                {true, true} ->
+                    throw({error, both_password_and_password_hash_are_provided});
+                %% clear password, update tags if needed
+                _ ->
+                    rabbit_auth_backend_internal:set_tags(Username, Tags, ActingUser),
+                    rabbit_auth_backend_internal:clear_password(Username, ActingUser)
+            end;
+        false ->
+            case {HasPassword, HasPasswordHash} of
+                {true, false}  ->
+                    create_user_with_password(PassedCredentialValidation, Username, Password, Tags, Permissions, ActingUser);
+                {false, true}  ->
+                    create_user_with_password_hash(Username, PasswordHash, Tags, User, Version, Permissions, ActingUser);
+                {true, true}   ->
+                    throw({error, both_password_and_password_hash_are_provided});
+                {false, false} ->
+                    %% this user won't be able to sign in using
+                    %% a username/password pair but can be used for x509 certificate authentication,
+                    %% with authn backends such as HTTP or LDAP and so on.
+                    create_user_with_password(PassedCredentialValidation, Username, <<"">>, Tags, Permissions, ActingUser)
+            end
+    end.
+
+update_user_password(_PassedCredentialValidation = true,  Username, Password, Tags, ActingUser) ->
+    rabbit_auth_backend_internal:change_password(Username, Password, ActingUser),
+    rabbit_auth_backend_internal:set_tags(Username, Tags, ActingUser);
+update_user_password(_PassedCredentialValidation = false, _Username, _Password, _Tags, _ActingUser) ->
+    %% we don't log here because
+    %% rabbit_auth_backend_internal will do it
+    throw({error, credential_validation_failed}).
+
+update_user_password_hash(Username, PasswordHash, Tags, User, Version, ActingUser) ->
+    %% when a hash this provided, credential validation
+    %% is not applied
+    HashingAlgorithm = hashing_algorithm(User, Version),
+
+    Hash = rabbit_misc:b64decode_or_throw(PasswordHash),
+    rabbit_auth_backend_internal:change_password_hash(
+      Username, Hash, HashingAlgorithm),
+    rabbit_auth_backend_internal:set_tags(Username, Tags, ActingUser).
+
+create_user_with_password(_PassedCredentialValidation = true,  Username, Password, Tags, undefined, ActingUser) ->
+    rabbit_auth_backend_internal:add_user(Username, Password, ActingUser),
+    rabbit_auth_backend_internal:set_tags(Username, Tags, ActingUser);
+create_user_with_password(_PassedCredentialValidation = true,  Username, Password, Tags, PreconfiguredPermissions, ActingUser) ->
+    rabbit_auth_backend_internal:add_user(Username, Password, ActingUser),
+    rabbit_auth_backend_internal:set_tags(Username, Tags, ActingUser),
+    preconfigure_permissions(Username, PreconfiguredPermissions, ActingUser);
+create_user_with_password(_PassedCredentialValidation = false, _Username, _Password, _Tags, _, _) ->
+    %% we don't log here because
+    %% rabbit_auth_backend_internal will do it
+    throw({error, credential_validation_failed}).
+
+create_user_with_password_hash(Username, PasswordHash, Tags, User, Version, PreconfiguredPermissions, ActingUser) ->
+    %% when a hash this provided, credential validation
+    %% is not applied
+    HashingAlgorithm = hashing_algorithm(User, Version),
+    Hash             = rabbit_misc:b64decode_or_throw(PasswordHash),
+
+    %% first we create a user with dummy credentials and no
+    %% validation applied, then we update password hash
+    TmpPassword = rabbit_guid:binary(rabbit_guid:gen_secure(), "tmp"),
+    rabbit_auth_backend_internal:add_user_sans_validation(Username, TmpPassword, ActingUser),
+
+    rabbit_auth_backend_internal:change_password_hash(
+      Username, Hash, HashingAlgorithm),
+    rabbit_auth_backend_internal:set_tags(Username, Tags, ActingUser),
+    preconfigure_permissions(Username, PreconfiguredPermissions, ActingUser).
+
+preconfigure_permissions(_Username, undefined, _ActingUser) ->
+    ok;
+preconfigure_permissions(Username, Map, ActingUser) when is_map(Map) ->
+    maps:map(fun(VHost, M) ->
+                     rabbit_auth_backend_internal:set_permissions(Username, VHost,
+                                                  maps:get(<<"configure">>, M),
+                                                  maps:get(<<"write">>,     M),
+                                                  maps:get(<<"read">>,      M),
+                                                  ActingUser)
+             end,
+             Map),
+    ok.
+
 %%----------------------------------------------------------------------------
 %% Listing
 
@@ -649,3 +781,21 @@ extract_topic_permission_params(Keys, #topic_permission{
         {exchange,  Exchange},
         {write,     WritePerm},
         {read,      ReadPerm}]).
+
+hashing_algorithm(User, Version) ->
+    case maps:get(hashing_algorithm, User, undefined) of
+        undefined ->
+            case Version of
+                %% 3.6.1 and later versions are supposed to have
+                %% the algorithm exported and thus not need a default
+                <<"3.6.0">>          -> rabbit_password_hashing_sha256;
+                <<"3.5.", _/binary>> -> rabbit_password_hashing_md5;
+                <<"3.4.", _/binary>> -> rabbit_password_hashing_md5;
+                <<"3.3.", _/binary>> -> rabbit_password_hashing_md5;
+                <<"3.2.", _/binary>> -> rabbit_password_hashing_md5;
+                <<"3.1.", _/binary>> -> rabbit_password_hashing_md5;
+                <<"3.0.", _/binary>> -> rabbit_password_hashing_md5;
+                _                    -> rabbit_password:hashing_mod()
+            end;
+        Alg       -> binary_to_atom(Alg, utf8)
+    end.
diff --git a/src/rabbit_definitions.erl b/src/rabbit_definitions.erl
new file mode 100644
index 0000000000..7961f6c3c6
--- /dev/null
+++ b/src/rabbit_definitions.erl
@@ -0,0 +1,422 @@
+%% 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 https://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 Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2019 Pivotal Software, Inc.  All rights reserved.
+%%
+
+-module(rabbit_definitions).
+-include_lib("rabbit_common/include/rabbit.hrl").
+
+-export([maybe_load_definitions/0, maybe_load_definitions_from/2]).
+-export([apply_defs/4, apply_defs/5]).
+-export([decode/1, decode/2, args/1]).
+
+maybe_load_definitions() ->
+    case application:get_env(rabbit, load_definitions) of
+        undefined  -> ok;
+        {ok, none} -> ok;
+        {ok, FileOrDir} ->
+            IsDir = filelib:is_dir(FileOrDir),
+            maybe_load_definitions_from(IsDir, FileOrDir)
+    end.
+
+maybe_load_definitions_from(true, Dir) ->
+    rabbit_log:info("Applying definitions from directory ~s", [Dir]),
+    load_definitions_from_files(file:list_dir(Dir), Dir);
+maybe_load_definitions_from(false, File) ->
+    load_definitions_from_file(File).
+
+load_definitions_from_files({ok, Filenames0}, Dir) ->
+    Filenames1 = lists:sort(Filenames0),
+    Filenames2 = [filename:join(Dir, F) || F <- Filenames1],
+    load_definitions_from_filenames(Filenames2);
+load_definitions_from_files({error, E}, Dir) ->
+    rabbit_log:error("Could not read definitions from directory ~s, Error: ~p", [Dir, E]),
+    {error, {could_not_read_defs, E}}.
+
+load_definitions_from_filenames([]) ->
+    ok;
+load_definitions_from_filenames([File|Rest]) ->
+    case load_definitions_from_file(File) of
+        ok         -> load_definitions_from_filenames(Rest);
+        {error, E} -> {error, {failed_to_import_definitions, File, E}}
+    end.
+
+load_definitions_from_file(File) ->
+    case file:read_file(File) of
+        {ok, Body} ->
+            rabbit_log:info("Applying definitions from ~s", [File]),
+            load_definitions(Body);
+        {error, E} ->
+            rabbit_log:error("Could not read definitions from ~s, Error: ~p", [File, E]),
+            {error, {could_not_read_defs, {File, E}}}
+    end.
+
+load_definitions(Body) ->
+    apply_defs(
+        Body, ?INTERNAL_USER,
+        fun () -> ok end,
+        fun (E) -> {error, E} end).
+
+decode(Keys, Body) ->
+    case decode(Body) of
+        {ok, J0} ->
+                    J = maps:fold(fun(K, V, Acc) ->
+                        Acc#{binary_to_atom(K, utf8) => V}
+                    end, J0, J0),
+                    Results = [get_or_missing(K, J) || K <- Keys],
+                    case [E || E = {key_missing, _} <- Results] of
+                        []      -> {ok, Results, J};
+                        Errors  -> {error, Errors}
+                    end;
+        Else     -> Else
+    end.
+
+decode(<<"">>) ->
+    {ok, #{}};
+decode(Body) ->
+    try
+        {ok, rabbit_json:decode(Body)}
+    catch error:_ -> {error, not_json}
+    end.
+
+apply_defs(Body, ActingUser, SuccessFun, ErrorFun) ->
+    rabbit_log:info("Asked to import definitions. Acting user: ~s", [rabbit_data_coercion:to_binary(ActingUser)]),
+    case decode([], Body) of
+        {error, E} ->
+            ErrorFun(E);
+        {ok, _, All} ->
+            Version = maps:get(rabbit_version, All, undefined),
+            try
+                rabbit_log:info("Importing users..."),
+                for_all(users,              ActingUser, All,
+                        fun(User, _Username) ->
+                                rabbit_auth_backend_internal:put_user(
+                                  User,
+                                  Version,
+                                  ActingUser)
+                        end),
+                rabbit_log:info("Importing vhosts..."),
+                for_all(vhosts,             ActingUser, All, fun add_vhost/2),
+                validate_limits(All),
+                rabbit_log:info("Importing user permissions..."),
+                for_all(permissions,        ActingUser, All, fun add_permission/2),
+                rabbit_log:info("Importing topic permissions..."),
+                for_all(topic_permissions,  ActingUser, All, fun add_topic_permission/2),
+                rabbit_log:info("Importing parameters..."),
+                for_all(parameters,         ActingUser, All, fun add_parameter/2),
+                rabbit_log:info("Importing global parameters..."),
+                for_all(global_parameters,  ActingUser, All, fun add_global_parameter/2),
+                rabbit_log:info("Importing policies..."),
+                for_all(policies,           ActingUser, All, fun add_policy/2),
+                rabbit_log:info("Importing queues..."),
+                for_all(queues,             ActingUser, All, fun add_queue/2),
+                rabbit_log:info("Importing exchanges..."),
+                for_all(exchanges,          ActingUser, All, fun add_exchange/2),
+                rabbit_log:info("Importing bindings..."),
+                for_all(bindings,           ActingUser, All, fun add_binding/2),
+                SuccessFun()
+            catch {error, E} -> ErrorFun(format(E));
+                  exit:E     -> ErrorFun(format(E))
+            end
+    end.
+
+apply_defs(Body, ActingUser, SuccessFun, ErrorFun, VHost) ->
+    rabbit_log:info("Asked to import definitions for a virtual host. Virtual host: ~p, acting user: ~p",
+                    [VHost, ActingUser]),
+    case decode([], Body) of
+        {error, E} ->
+            ErrorFun(E);
+        {ok, _, All} ->
+            try
+                validate_limits(All, VHost),
+                rabbit_log:info("Importing parameters..."),
+                for_all(parameters,  ActingUser, All, VHost, fun add_parameter/3),
+                rabbit_log:info("Importing policies..."),
+                for_all(policies,    ActingUser, All, VHost, fun add_policy/3),
+                rabbit_log:info("Importing queues..."),
+                for_all(queues,      ActingUser, All, VHost, fun add_queue/3),
+                rabbit_log:info("Importing exchanges..."),
+                for_all(exchanges,   ActingUser, All, VHost, fun add_exchange/3),
+                rabbit_log:info("Importing bindings..."),
+                for_all(bindings,    ActingUser, All, VHost, fun add_binding/3),
+                SuccessFun()
+            catch {error, E} -> ErrorFun(format(E));
+                  exit:E     -> ErrorFun(format(E))
+            end
+    end.
+
+for_all(Name, ActingUser, All, Fun) ->
+    case maps:get(Name, All, undefined) of
+        undefined -> ok;
+        List      -> [Fun(maps:from_list([{atomise_name(K), V} || {K, V} <- maps:to_list(M)]),
+                          ActingUser) ||
+                         M <- List, is_map(M)]
+    end.
+
+for_all(Name, ActingUser, All, VHost, Fun) ->
+    case maps:get(Name, All, undefined) of
+        undefined -> ok;
+        List      -> [Fun(VHost, maps:from_list([{atomise_name(K), V} || {K, V} <- maps:to_list(M)]),
+                          ActingUser) ||
+                         M <- List, is_map(M)]
+    end.
+
+format(#amqp_error{name = Name, explanation = Explanation}) ->
+    rabbit_data_coercion:to_binary(rabbit_misc:format("~s: ~s", [Name, Explanation]));
+format({no_such_vhost, undefined}) ->
+    rabbit_data_coercion:to_binary(
+      "Virtual host does not exist and is not specified in definitions file.");
+format({no_such_vhost, VHost}) ->
+    rabbit_data_coercion:to_binary(
+      rabbit_misc:format("Please create virtual host \"~s\" prior to importing definitions.",
+                         [VHost]));
+format({vhost_limit_exceeded, ErrMsg}) ->
+    rabbit_data_coercion:to_binary(ErrMsg);
+format(E) ->
+    rabbit_data_coercion:to_binary(rabbit_misc:format("~p", [E])).
+
+add_parameter(Param, Username) ->
+    VHost = maps:get(vhost,     Param, undefined),
+    add_parameter(VHost, Param, Username).
+
+add_parameter(VHost, Param, Username) ->
+    Comp  = maps:get(component, Param, undefined),
+    Key   = maps:get(name,      Param, undefined),
+    Term  = maps:get(value,     Param, undefined),
+    Result = case is_map(Term) of
+        true ->
+            %% coerce maps to proplists for backwards compatibility.
+            %% See rabbitmq-management#528.
+            TermProplist = rabbit_data_coercion:to_proplist(Term),
+            rabbit_runtime_parameters:set(VHost, Comp, Key, TermProplist, Username);
+        _ ->
+            rabbit_runtime_parameters:set(VHost, Comp, Key, Term, Username)
+    end,
+    case Result of
+        ok                -> ok;
+        {error_string, E} ->
+            S = rabbit_misc:format(" (~s/~s/~s)", [VHost, Comp, Key]),
+            exit(rabbit_data_coercion:to_binary(rabbit_misc:escape_html_tags(E ++ S)))
+    end.
+
+add_global_parameter(Param, Username) ->
+    Key   = maps:get(name,      Param, undefined),
+    Term  = maps:get(value,     Param, undefined),
+    case is_map(Term) of
+        true ->
+            %% coerce maps to proplists for backwards compatibility.
+            %% See rabbitmq-management#528.
+            TermProplist = rabbit_data_coercion:to_proplist(Term),
+            rabbit_runtime_parameters:set_global(Key, TermProplist, Username);
+        _ ->
+            rabbit_runtime_parameters:set_global(Key, Term, Username)
+    end.
+
+add_policy(Param, Username) ->
+    VHost = maps:get(vhost, Param, undefined),
+    add_policy(VHost, Param, Username).
+
+add_policy(VHost, Param, Username) ->
+    Key   = maps:get(name,  Param, undefined),
+    case rabbit_policy:set(
+           VHost, Key, maps:get(pattern, Param, undefined),
+           case maps:get(definition, Param, undefined) of
+               undefined -> undefined;
+               Def -> rabbit_data_coercion:to_proplist(Def)
+           end,
+           maps:get(priority, Param, undefined),
+           maps:get('apply-to', Param, <<"all">>),
+           Username) of
+        ok                -> ok;
+        {error_string, E} -> S = rabbit_misc:format(" (~s/~s)", [VHost, Key]),
+                             exit(rabbit_data_coercion:to_binary(rabbit_misc:escape_html_tags(E ++ S)))
+    end.
+
+add_vhost(VHost, ActingUser) ->
+    VHostName = maps:get(name, VHost, undefined),
+    VHostTrace = maps:get(tracing, VHost, undefined),
+    VHostDefinition = maps:get(definition, VHost, undefined),
+    VHostTags = maps:get(tags, VHost, undefined),
+    rabbit_vhost:put_vhost(VHostName, VHostDefinition, VHostTags, VHostTrace, ActingUser).
+
+add_permission(Permission, ActingUser) ->
+    rabbit_auth_backend_internal:set_permissions(maps:get(user,      Permission, undefined),
+                                                 maps:get(vhost,     Permission, undefined),
+                                                 maps:get(configure, Permission, undefined),
+                                                 maps:get(write,     Permission, undefined),
+                                                 maps:get(read,      Permission, undefined),
+                                                 ActingUser).
+
+add_topic_permission(TopicPermission, ActingUser) ->
+    rabbit_auth_backend_internal:set_topic_permissions(
+        maps:get(user,      TopicPermission, undefined),
+        maps:get(vhost,     TopicPermission, undefined),
+        maps:get(exchange,  TopicPermission, undefined),
+        maps:get(write,     TopicPermission, undefined),
+        maps:get(read,      TopicPermission, undefined),
+      ActingUser).
+
+add_queue(Queue, ActingUser) ->
+    add_queue_int(Queue, r(queue, Queue), ActingUser).
+
+add_queue(VHost, Queue, ActingUser) ->
+    add_queue_int(Queue, rv(VHost, queue, Queue), ActingUser).
+
+add_queue_int(Queue, Name, ActingUser) ->
+    rabbit_amqqueue:declare(Name,
+                            maps:get(durable,                         Queue, undefined),
+                            maps:get(auto_delete,                     Queue, undefined),
+                            args(maps:get(arguments, Queue, undefined)),
+                            none,
+                            ActingUser).
+
+add_exchange(Exchange, ActingUser) ->
+    add_exchange_int(Exchange, r(exchange, Exchange), ActingUser).
+
+add_exchange(VHost, Exchange, ActingUser) ->
+    add_exchange_int(Exchange, rv(VHost, exchange, Exchange), ActingUser).
+
+add_exchange_int(Exchange, Name, ActingUser) ->
+    Internal = case maps:get(internal, Exchange, undefined) of
+                   undefined -> false; %% =< 2.2.0
+                   I         -> I
+               end,
+    rabbit_exchange:declare(Name,
+                            rabbit_exchange:check_type(maps:get(type, Exchange, undefined)),
+                            maps:get(durable,                         Exchange, undefined),
+                            maps:get(auto_delete,                     Exchange, undefined),
+                            Internal,
+                            args(maps:get(arguments, Exchange, undefined)),
+                            ActingUser).
+
+add_binding(Binding, ActingUser) ->
+    DestType = dest_type(Binding),
+    add_binding_int(Binding, r(exchange, source, Binding),
+                    r(DestType, destination, Binding), ActingUser).
+
+add_binding(VHost, Binding, ActingUser) ->
+    DestType = dest_type(Binding),
+    add_binding_int(Binding, rv(VHost, exchange, source, Binding),
+                    rv(VHost, DestType, destination, Binding), ActingUser).
+
+add_binding_int(Binding, Source, Destination, ActingUser) ->
+    rabbit_binding:add(
+      #binding{source       = Source,
+               destination  = Destination,
+               key          = maps:get(routing_key, Binding, undefined),
+               args         = args(maps:get(arguments, Binding, undefined))},
+      ActingUser).
+
+dest_type(Binding) ->
+    list_to_atom(binary_to_list(maps:get(destination_type, Binding, undefined))).
+
+r(Type, Props) -> r(Type, name, Props).
+
+r(Type, Name, Props) ->
+    rabbit_misc:r(maps:get(vhost, Props, undefined), Type, maps:get(Name, Props, undefined)).
+
+rv(VHost, Type, Props) -> rv(VHost, Type, name, Props).
+
+rv(VHost, Type, Name, Props) ->
+    rabbit_misc:r(VHost, Type, maps:get(Name, Props, undefined)).
+
+%%--------------------------------------------------------------------
+
+validate_limits(All) ->
+    case maps:get(queues, All, undefined) of
+        undefined -> ok;
+        Queues0 ->
+            {ok, VHostMap} = filter_out_existing_queues(Queues0),
+            maps:fold(fun validate_vhost_limit/3, ok, VHostMap)
+    end.
+
+validate_limits(All, VHost) ->
+    case maps:get(queues, All, undefined) of
+        undefined -> ok;
+        Queues0 ->
+            Queues1 = filter_out_existing_queues(VHost, Queues0),
+            AddCount = length(Queues1),
+            validate_vhost_limit(VHost, AddCount, ok)
+    end.
+
+filter_out_existing_queues(Queues) ->
+    build_filtered_map(Queues, maps:new()).
+
+filter_out_existing_queues(VHost, Queues) ->
+    Pred = fun(Queue) ->
+               Rec = rv(VHost, queue, <<"name">>, Queue),
+               case rabbit_amqqueue:lookup(Rec) of
+                   {ok, _} -> false;
+                   {error, not_found} -> true
+               end
+           end,
+    lists:filter(Pred, Queues).
+
+build_queue_data(Queue) ->
+    VHost = maps:get(<<"vhost">>, Queue, undefined),
+    Rec = rv(VHost, queue, <<"name">>, Queue),
+    {Rec, VHost}.
+
+build_filtered_map([], AccMap) ->
+    {ok, AccMap};
+build_filtered_map([Queue|Rest], AccMap0) ->
+    {Rec, VHost} = build_queue_data(Queue),
+    case rabbit_amqqueue:lookup(Rec) of
+        {error, not_found} ->
+            AccMap1 = maps_update_with(VHost, fun(V) -> V + 1 end, 1, AccMap0),
+            build_filtered_map(Rest, AccMap1);
+        {ok, _} ->
+            build_filtered_map(Rest, AccMap0)
+    end.
+
+%% Copy of maps:with_util/3 from Erlang 20.0.1.
+maps_update_with(Key,Fun,Init,Map) when is_function(Fun,1), is_map(Map) ->
+    case maps:find(Key,Map) of
+        {ok,Val} -> maps:update(Key,Fun(Val),Map);
+        error -> maps:put(Key,Init,Map)
+    end;
+maps_update_with(Key,Fun,Init,Map) ->
+    erlang:error(maps_error_type(Map),[Key,Fun,Init,Map]).
+
+%% Copy of maps:error_type/1 from Erlang 20.0.1.
+maps_error_type(M) when is_map(M) -> badarg;
+maps_error_type(V) -> {badmap, V}.
+
+validate_vhost_limit(VHost, AddCount, ok) ->
+    WouldExceed = rabbit_vhost_limit:would_exceed_queue_limit(AddCount, VHost),
+    validate_vhost_queue_limit(VHost, AddCount, WouldExceed).
+
+validate_vhost_queue_limit(_VHost, 0, _) ->
+    % Note: not adding any new queues so the upload
+    % must be update-only
+    ok;
+validate_vhost_queue_limit(_VHost, _AddCount, false) ->
+    % Note: would not exceed queue limit
+    ok;
+validate_vhost_queue_limit(VHost, AddCount, {true, Limit, QueueCount}) ->
+    ErrFmt = "Adding ~B queue(s) to virtual host \"~s\" would exceed the limit of ~B queue(s).~n~nThis virtual host currently has ~B queue(s) defined.~n~nImport aborted!",
+    ErrInfo = [AddCount, VHost, Limit, QueueCount],
+    ErrMsg = rabbit_misc:format(ErrFmt, ErrInfo),
+    exit({vhost_limit_exceeded, ErrMsg}).
+
+atomise_name(N) -> rabbit_data_coercion:to_atom(N).
+
+get_or_missing(K, L) ->
+    case maps:get(K, L, undefined) of
+        undefined -> {key_missing, K};
+        V         -> V
+    end.
+
+args([]) -> args(#{});
+args(L)  -> rabbit_misc:to_amqp_table(L).
diff --git a/src/rabbit_vhost.erl b/src/rabbit_vhost.erl
index 95758cec7f..71cc7c93d2 100644
--- a/src/rabbit_vhost.erl
+++ b/src/rabbit_vhost.erl
@@ -27,6 +27,7 @@
 -export([dir/1, msg_store_dir_path/1, msg_store_dir_wildcard/0]).
 -export([delete_storage/1]).
 -export([vhost_down/1]).
+-export([put_vhost/5]).
 
 %%
 %% API
@@ -171,6 +172,48 @@ delete(VHost, ActingUser) ->
     rabbit_vhost_sup_sup:delete_on_all_nodes(VHost),
     ok.
 
+put_vhost(Name, Description, Tags0, Trace, Username) ->
+    Tags = case Tags0 of
+      undefined   -> <<"">>;
+      null        -> <<"">>;
+      "undefined" -> <<"">>;
+      "null"      -> <<"">>;
+      Other       -> Other
+    end,
+    Result = case exists(Name) of
+        true  -> ok;
+        false -> add(Name, Description, parse_tags(Tags), Username),
+                 %% wait for up to 45 seconds for the vhost to initialise
+                 %% on all nodes
+                 case await_running_on_all_nodes(Name, 45000) of
+                     ok               ->
+                         maybe_grant_full_permissions(Name, Username);
+                     {error, timeout} ->
+                         {error, timeout}
+                 end
+    end,
+    case Trace of
+        true      -> rabbit_trace:start(Name);
+        false     -> rabbit_trace:stop(Name);
+        undefined -> ok
+    end,
+    Result.
+
+%% when definitions are loaded on boot, Username here will be ?INTERNAL_USER,
+%% which does not actually exist
+maybe_grant_full_permissions(_Name, ?INTERNAL_USER) ->
+    ok;
+maybe_grant_full_permissions(Name, Username) ->
+    U = rabbit_auth_backend_internal:lookup_user(Username),
+    maybe_grant_full_permissions(U, Name, Username).
+
+maybe_grant_full_permissions({ok, _}, Name, Username) ->
+    rabbit_auth_backend_internal:set_permissions(
+      Username, Name, <<".*">>, <<".*">>, <<".*">>, Username);
+maybe_grant_full_permissions(_, _Name, _Username) ->
+    ok.
+
+
 %% 50 ms
 -define(AWAIT_SAMPLE_INTERVAL, 50).