Import definitions: support user limits

5 files changed, 225 insertions, 61 deletions

diff --git a/deps/rabbit/src/rabbit_auth_backend_internal.erl b/deps/rabbit/src/rabbit_auth_backend_internal.erl
index 06467d9633..f9b2dd893c 100644
--- a/deps/rabbit/src/rabbit_auth_backend_internal.erl
+++ b/deps/rabbit/src/rabbit_auth_backend_internal.erl
@@ -20,9 +20,9 @@
          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, put_user/2, put_user/3,
-         change_password_and_tags/4,
-         change_password_hash_and_tags/4,
-         add_user_sans_validation/5]).
+         update_user/5,
+         update_user_with_hash/5,
+         add_user_sans_validation/6]).
 
 -export([set_user_limits/3, clear_user_limits/3, is_over_connection_limit/1,
          is_over_channel_limit/1, get_user_limits/0, get_user_limits/1]).
@@ -215,18 +215,21 @@ add_user(Username, Password, ActingUser) ->
                rabbit_types:username(), [atom()]) -> 'ok' | {'error', string()}.
 
 add_user(Username, Password, ActingUser, Tags) ->
+    add_user(Username, Password, ActingUser, undefined, Tags).
+
+add_user(Username, Password, ActingUser, Limits, Tags) ->
     validate_and_alternate_credentials(Username, Password, ActingUser,
-                                       add_user_sans_validation(Tags)).
+                                       add_user_sans_validation(Limits, Tags)).
 
 add_user_sans_validation(Username, Password, ActingUser) ->
-    add_user_sans_validation(Username, Password, ActingUser, []).
+    add_user_sans_validation(Username, Password, ActingUser, undefined, []).
 
-add_user_sans_validation(Tags) ->
+add_user_sans_validation(Limits, Tags) ->
     fun(Username, Password, ActingUser) ->
-            add_user_sans_validation(Username, Password, ActingUser, Tags)
+            add_user_sans_validation(Username, Password, ActingUser, Limits, Tags)
     end.
 
-add_user_sans_validation(Username, Password, ActingUser, Tags) ->
+add_user_sans_validation(Username, Password, ActingUser, Limits, Tags) ->
     rabbit_log:debug("Asked to create a new user '~s', password length in bytes: ~p", [Username, bit_size(Password)]),
     %% hash_password will pick the hashing function configured for us
     %% but we also need to store a hint as part of the record, so we
@@ -235,21 +238,29 @@ add_user_sans_validation(Username, Password, ActingUser, Tags) ->
     PasswordHash = hash_password(HashingMod, Password),
     User0 = internal_user:create_user(Username, PasswordHash, HashingMod),
     ConvertedTags = [rabbit_data_coercion:to_atom(I) || I <- Tags],
-    User = internal_user:set_tags(User0, ConvertedTags),
-    add_user_sans_validation_in(Username, User, ConvertedTags, ActingUser).
-
-add_user_sans_validation(Username, PasswordHash, HashingAlgorithm, Tags, ActingUser) ->
+    User1 = internal_user:set_tags(User0, ConvertedTags),
+    User = case Limits of
+               undefined -> User1;
+               Term -> internal_user:update_limits(add, User1, Term)
+           end,
+    add_user_sans_validation_in(Username, User, ConvertedTags, Limits, ActingUser).
+
+add_user_sans_validation(Username, PasswordHash, HashingAlgorithm, Tags, Limits, ActingUser) ->
     rabbit_log:debug("Asked to create a new user '~s' with password hash", [Username]),
     ConvertedTags = [rabbit_data_coercion:to_atom(I) || I <- Tags],
     HashingMod = rabbit_password:hashing_mod(),
     User0 = internal_user:create_user(Username, PasswordHash, HashingMod),
-    User = internal_user:set_tags(
-             internal_user:set_password_hash(User0,
-                                             PasswordHash, HashingAlgorithm),
-             ConvertedTags),
-    add_user_sans_validation_in(Username, User, ConvertedTags, ActingUser).
-
-add_user_sans_validation_in(Username, User, ConvertedTags, ActingUser) ->
+    User1 = internal_user:set_tags(
+              internal_user:set_password_hash(User0,
+                                              PasswordHash, HashingAlgorithm),
+              ConvertedTags),
+    User = case Limits of
+               undefined -> User1;
+               Term -> internal_user:update_limits(add, User1, Term)
+           end,
+    add_user_sans_validation_in(Username, User, ConvertedTags, Limits, ActingUser).
+
+add_user_sans_validation_in(Username, User, ConvertedTags, Limits, ActingUser) ->
     try
         R = rabbit_misc:execute_mnesia_transaction(
           fun () ->
@@ -267,6 +278,10 @@ add_user_sans_validation_in(Username, User, ConvertedTags, ActingUser) ->
             [] -> ok;
             _ -> notify_user_tags_set(Username, ConvertedTags, ActingUser)
         end,
+        case Limits of
+            undefined -> ok;
+            _ -> notify_limit_set(Username, ActingUser, Limits)
+        end,
         R
     catch
         throw:{error, {user_already_exists, _}} = Error ->
@@ -360,11 +375,11 @@ change_password_sans_validation(Username, Password, ActingUser) ->
             erlang:raise(Class, Error, Stacktrace)
     end.
 
-change_password_and_tags(Username, Password, Tags, ActingUser) ->
+update_user(Username, Password, Tags, Limits, ActingUser) ->
     validate_and_alternate_credentials(Username, Password, ActingUser,
-                                       change_password_and_tags_sans_validation(Tags)).
+                                       update_user_sans_validation(Tags, Limits)).
 
-change_password_and_tags_sans_validation(Tags) ->
+update_user_sans_validation(Tags, Limits) ->
     fun(Username, Password, ActingUser) ->
             try
                 rabbit_log:debug("Asked to change password of user '~s', new password length in bytes: ~p", [Username, bit_size(Password)]),
@@ -373,11 +388,12 @@ change_password_and_tags_sans_validation(Tags) ->
                 rabbit_log:debug("Asked to set user tags for user '~s' to ~p", [Username, Tags]),
 
                 ConvertedTags = [rabbit_data_coercion:to_atom(I) || I <- Tags],
-                R = change_password_hash_and_tags(Username,
-                                                  hash_password(rabbit_password:hashing_mod(),
-                                                                Password),
-                                                  HashingAlgorithm,
-                                                  ConvertedTags),
+                R = update_user_with_hash(Username,
+                                          hash_password(rabbit_password:hashing_mod(),
+                                                        Password),
+                                          HashingAlgorithm,
+                                          ConvertedTags,
+                                          Limits),
                 rabbit_log:info("Successfully changed password for user '~s'", [Username]),
                 rabbit_event:notify(user_password_changed,
                                     [{name, Username},
@@ -419,18 +435,22 @@ change_password_hash(Username, PasswordHash) ->
 
 
 change_password_hash(Username, PasswordHash, HashingAlgorithm) ->
-    update_user(Username, fun(User) ->
-                              internal_user:set_password_hash(User,
-                                  PasswordHash, HashingAlgorithm)
-                          end).
-
-change_password_hash_and_tags(Username, PasswordHash, HashingAlgorithm, ConvertedTags) ->
-    update_user(Username, fun(User) ->
-                                  internal_user:set_tags(
-                                    internal_user:set_password_hash(User,
-                                                                    PasswordHash, HashingAlgorithm),
-                                    ConvertedTags)
-                          end).
+    update_user_with_hash(Username, PasswordHash, HashingAlgorithm, [], undefined).
+
+update_user_with_hash(Username, PasswordHash, HashingAlgorithm, ConvertedTags, Limits) ->
+    update_user(Username,
+                fun(User0) ->
+                        User1 = internal_user:set_password_hash(User0,
+                                                                PasswordHash, HashingAlgorithm),
+                        User2 = case Limits of
+                                    undefined -> User1;
+                                    _         -> internal_user:update_limits(add, User1, Limits)
+                                end,
+                        case ConvertedTags of
+                            [] -> User2;
+                            _  -> internal_user:set_tags(User2, ConvertedTags)
+                        end
+                end).
 
 -spec set_tags(rabbit_types:username(), [atom()], rabbit_types:username()) -> 'ok'.
 
@@ -732,13 +752,27 @@ put_user(User, Version, ActingUser) ->
                 rabbit_credential_validation:validate(Username, Password) =:= ok
         end,
 
+    Limits = case rabbit_feature_flags:is_enabled(user_limits) of
+                 false ->
+                     undefined;
+                 true ->
+                     case maps:get(limits, User, undefined) of
+                         undefined ->
+                             undefined;
+                         Term ->
+                             case validate_user_limits(Term) of
+                                 ok -> Term;
+                                 Error -> throw(Error)
+                             end
+                     end
+             end,
     case exists(Username) of
         true  ->
             case {HasPassword, HasPasswordHash} of
                 {true, false} ->
-                    update_user_password(PassedCredentialValidation, Username, Password, Tags, ActingUser);
+                    update_user_password(PassedCredentialValidation, Username, Password, Tags, Limits, ActingUser);
                 {false, true} ->
-                    update_user_password_hash(Username, PasswordHash, Tags, User, Version);
+                    update_user_password_hash(Username, PasswordHash, Tags, Limits, User, Version);
                 {true, true} ->
                     throw({error, both_password_and_password_hash_are_provided});
                 %% clear password, update tags if needed
@@ -749,54 +783,54 @@ put_user(User, Version, ActingUser) ->
         false ->
             case {HasPassword, HasPasswordHash} of
                 {true, false}  ->
-                    create_user_with_password(PassedCredentialValidation, Username, Password, Tags, Permissions, ActingUser);
+                    create_user_with_password(PassedCredentialValidation, Username, Password, Tags, Permissions, Limits, ActingUser);
                 {false, true}  ->
-                    create_user_with_password_hash(Username, PasswordHash, Tags, User, Version, Permissions, ActingUser);
+                    create_user_with_password_hash(Username, PasswordHash, Tags, User, Version, Permissions, Limits, 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)
+                    create_user_with_password(PassedCredentialValidation, Username, <<"">>, Tags, Permissions, Limits, ActingUser)
             end
     end.
 
-update_user_password(_PassedCredentialValidation = true,  Username, Password, Tags, ActingUser) ->
-    %% change_password, set_tags
-    rabbit_auth_backend_internal:change_password_and_tags(Username, Password, Tags, ActingUser);
-update_user_password(_PassedCredentialValidation = false, _Username, _Password, _Tags, _ActingUser) ->
+update_user_password(_PassedCredentialValidation = true,  Username, Password, Tags, Limits, ActingUser) ->
+    %% change_password, set_tags and limits
+    rabbit_auth_backend_internal:update_user(Username, Password, Tags, Limits, ActingUser);
+update_user_password(_PassedCredentialValidation = false, _Username, _Password, _Tags, _Limits, _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) ->
+update_user_password_hash(Username, PasswordHash, Tags, Limits, User, Version) ->
     %% when a hash this provided, credential validation
     %% is not applied
     HashingAlgorithm = hashing_algorithm(User, Version),
 
     Hash = rabbit_misc:b64decode_or_throw(PasswordHash),
     ConvertedTags = [rabbit_data_coercion:to_atom(I) || I <- Tags],
-    rabbit_auth_backend_internal:change_password_hash_and_tags(
-      Username, Hash, HashingAlgorithm, ConvertedTags).
+    rabbit_auth_backend_internal:update_user_with_hash(
+      Username, Hash, HashingAlgorithm, ConvertedTags, Limits).
 
-create_user_with_password(_PassedCredentialValidation = true,  Username, Password, Tags, undefined, ActingUser) ->
-    rabbit_auth_backend_internal:add_user(Username, Password, ActingUser, Tags);
-create_user_with_password(_PassedCredentialValidation = true,  Username, Password, Tags, PreconfiguredPermissions, ActingUser) ->
-    rabbit_auth_backend_internal:add_user(Username, Password, ActingUser, Tags),
+create_user_with_password(_PassedCredentialValidation = true,  Username, Password, Tags, undefined, Limits, ActingUser) ->
+    rabbit_auth_backend_internal:add_user(Username, Password, ActingUser, Limits, Tags);
+create_user_with_password(_PassedCredentialValidation = true,  Username, Password, Tags, PreconfiguredPermissions, Limits, ActingUser) ->
+    rabbit_auth_backend_internal:add_user(Username, Password, ActingUser, Limits, Tags),
     preconfigure_permissions(Username, PreconfiguredPermissions, ActingUser);
-create_user_with_password(_PassedCredentialValidation = false, _Username, _Password, _Tags, _, _) ->
+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) ->
+create_user_with_password_hash(Username, PasswordHash, Tags, User, Version, PreconfiguredPermissions, Limits, 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:add_user_sans_validation(Username, Hash, HashingAlgorithm, Tags, ActingUser),
+    rabbit_auth_backend_internal:add_user_sans_validation(Username, Hash, HashingAlgorithm, Tags, Limits, ActingUser),
     preconfigure_permissions(Username, PreconfiguredPermissions, ActingUser).
 
 preconfigure_permissions(_Username, undefined, _ActingUser) ->
@@ -831,8 +865,7 @@ set_user_limits(Username, Definition, ActingUser) when is_map(Definition) ->
     end.
 
 validate_parameters_and_update_limit(Username, Term, ActingUser) ->
-    case flatten_errors(rabbit_parameter_validation:proplist(
-                        <<"user-limits">>, user_limit_validation(), Term)) of
+    case validate_user_limits(Term) of
         ok ->
             update_user(Username, fun(User) ->
                                       internal_user:update_limits(add, User, Term)
@@ -842,6 +875,10 @@ validate_parameters_and_update_limit(Username, Term, ActingUser) ->
             {error_string, rabbit_misc:format(Reason, Arguments)}
     end.
 
+validate_user_limits(Term) ->
+    flatten_errors(rabbit_parameter_validation:proplist(
+                     <<"user-limits">>, user_limit_validation(), Term)).
+
 user_limit_validation() ->
     [{<<"max-connections">>, fun rabbit_parameter_validation:integer/2, optional},
      {<<"max-channels">>, fun rabbit_parameter_validation:integer/2, optional}].
diff --git a/deps/rabbit/src/rabbit_osiris_metrics.erl b/deps/rabbit/src/rabbit_osiris_metrics.erl
index e93d81dc47..710ce1b65e 100644
--- a/deps/rabbit/src/rabbit_osiris_metrics.erl
+++ b/deps/rabbit/src/rabbit_osiris_metrics.erl
@@ -78,6 +78,8 @@ handle_info(tick, #state{timeout = Timeout} = State) ->
                               %% down `rabbit_sup` and the whole `rabbit` app.
                               []
                       end,
+
+
               rabbit_core_metrics:queue_stats(QName, Infos),
               rabbit_event:notify(queue_stats, Infos ++ [{name, QName},
                                                          {messages, COffs},
diff --git a/deps/rabbit/test/definition_import_SUITE.erl b/deps/rabbit/test/definition_import_SUITE.erl
index 3685a2addd..ba3cb979d1 100644
--- a/deps/rabbit/test/definition_import_SUITE.erl
+++ b/deps/rabbit/test/definition_import_SUITE.erl
@@ -47,7 +47,9 @@ groups() ->
                                import_case14,
                                import_case15,
                                import_case16,
-                               import_case17
+                               import_case17,
+                               import_case18,
+                               import_case19
                               ]},
         
         {boot_time_import_using_classic_source, [], [
@@ -239,6 +241,34 @@ import_case16(Config) ->
 
 import_case17(Config) -> import_invalid_file_case(Config, "failing_case17").
 
+import_case18(Config) ->
+    case rabbit_ct_helpers:is_mixed_versions() of
+      false ->
+        case rabbit_ct_broker_helpers:enable_feature_flag(Config, user_limits) of
+            ok ->
+                import_file_case(Config, "case18"),
+                User = <<"limited_guest">>,
+                UserIsImported =
+                    fun () ->
+                            case user_lookup(Config, User) of
+                                {error, not_found} -> false;
+                                _       -> true
+                            end
+                    end,
+                rabbit_ct_helpers:await_condition(UserIsImported, 20000),
+                {ok, UserRec} = user_lookup(Config, User),
+                ?assertEqual(#{<<"max-connections">> => 2}, internal_user:get_limits(UserRec)),
+                ok;
+            Skip ->
+                Skip
+        end;
+      _ ->
+        %% skip the test in mixed version mode
+        {skip, "Should not run in mixed version environments"}
+    end.
+
+import_case19(Config) -> import_invalid_file_case(Config, "failing_case19").
+
 export_import_round_trip_case1(Config) ->
     case rabbit_ct_helpers:is_mixed_versions() of
       false ->
@@ -385,3 +415,6 @@ queue_lookup(Config, VHost, Name) ->
 
 vhost_lookup(Config, VHost) ->
     rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_vhost, lookup, [VHost]).
+
+user_lookup(Config, User) ->
+    rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_auth_backend_internal, lookup_user, [User]).
diff --git a/deps/rabbit/test/definition_import_SUITE_data/case18.json b/deps/rabbit/test/definition_import_SUITE_data/case18.json
new file mode 100644
index 0000000000..9e0f755beb
--- /dev/null
+++ b/deps/rabbit/test/definition_import_SUITE_data/case18.json
@@ -0,0 +1,46 @@
+{
+  "bindings": [],
+  "exchanges": [],
+  "global_parameters": [
+    {
+      "name": "cluster_name",
+      "value": "rabbitmq@localhost"
+    }
+  ],
+  "parameters": [],
+  "permissions": [
+    {
+      "configure": ".*",
+      "read": ".*",
+      "user": "guest",
+      "vhost": "/",
+      "write": ".*"
+    }
+  ],
+  "policies": [],
+  "queues": [],
+  "rabbit_version": "3.9.1",
+  "rabbitmq_version": "3.9.1",
+  "topic_permissions": [],
+  "users": [
+    {
+      "hashing_algorithm": "rabbit_password_hashing_sha256",
+      "limits": {"max-connections" : 2},
+      "name": "limited_guest",
+      "password_hash": "wS4AT3B4Z5RpWlFn1FA30osf2C75D7WA3gem591ACDZ6saO6",
+      "tags": [
+        "administrator"
+      ]
+    }
+  ],
+  "vhosts": [
+    {
+      "limits": [],
+      "name": "/"
+    },
+    {
+      "limits": [],
+      "name": "tagged"
+    }
+  ]
+}
diff --git a/deps/rabbit/test/definition_import_SUITE_data/failing_case19.json b/deps/rabbit/test/definition_import_SUITE_data/failing_case19.json
new file mode 100644
index 0000000000..ab9d355538
--- /dev/null
+++ b/deps/rabbit/test/definition_import_SUITE_data/failing_case19.json
@@ -0,0 +1,46 @@
+{
+  "bindings": [],
+  "exchanges": [],
+  "global_parameters": [
+    {
+      "name": "cluster_name",
+      "value": "rabbitmq@localhost"
+    }
+  ],
+  "parameters": [],
+  "permissions": [
+    {
+      "configure": ".*",
+      "read": ".*",
+      "user": "guest",
+      "vhost": "/",
+      "write": ".*"
+    }
+  ],
+  "policies": [],
+  "queues": [],
+  "rabbit_version": "3.9.1",
+  "rabbitmq_version": "3.9.1",
+  "topic_permissions": [],
+  "users": [
+    {
+      "hashing_algorithm": "rabbit_password_hashing_sha256",
+      "limits": {"max-connections" : "twomincepies"},
+      "name": "limited_guest",
+      "password_hash": "wS4AT3B4Z5RpWlFn1FA30osf2C75D7WA3gem591ACDZ6saO6",
+      "tags": [
+        "administrator"
+      ]
+    }
+  ],
+  "vhosts": [
+    {
+      "limits": [],
+      "name": "/"
+    },
+    {
+      "limits": [],
+      "name": "tagged"
+    }
+  ]
+}