diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/rabbit.erl | 9 | ||||
| -rw-r--r-- | src/rabbit_auth_backend_internal.erl | 152 | ||||
| -rw-r--r-- | src/rabbit_definitions.erl | 422 | ||||
| -rw-r--r-- | src/rabbit_vhost.erl | 43 |
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). |