Notify of permission deletions when deleting a vhost

the-mikedavis · the-mikedavis · commit d8fcfc67617d · 2024-04-25T10:26:09.000-04:00
diff --git a/deps/rabbit/src/rabbit_auth_backend_internal.erl b/deps/rabbit/src/rabbit_auth_backend_internal.erl
@@ -17,8 +17,9 @@
 -export([add_user/3, add_user/4, add_user/5, delete_user/2, lookup_user/1, exists/1,
          change_password/3, clear_password/2,
          hash_password/2, change_password_hash/2, change_password_hash/3,
-         set_tags/3, set_permissions/6, clear_permissions/3, clear_permissions_for_vhost/2, set_permissions_globally/5,
-         set_topic_permissions/6, clear_topic_permissions/3, clear_topic_permissions/4, clear_topic_permissions_for_vhost/2,
+         set_tags/3, set_permissions/6, clear_permissions/3, set_permissions_globally/5,
+         set_topic_permissions/6, clear_topic_permissions/3, clear_topic_permissions/4,
+         clear_all_permissions_for_vhost/2,
          add_user_sans_validation/3, put_user/2, put_user/3,
          update_user/5,
          update_user_with_hash/5,
@@ -540,8 +541,35 @@ clear_permissions(Username, VirtualHost, ActingUser) ->
             erlang:raise(Class, Error, Stacktrace)
     end.
 
-clear_permissions_for_vhost(VirtualHost, _ActingUser) ->
-    rabbit_db_user:clear_matching_user_permissions('_', VirtualHost).
+-spec clear_all_permissions_for_vhost(VirtualHost, ActingUser) -> Ret when
+      VirtualHost :: rabbit_types:vhost(),
+      ActingUser :: rabbit_types:username(),
+      Ret :: ok | {error, Reason :: any()}.
+
+clear_all_permissions_for_vhost(VirtualHost, ActingUser) ->
+    case rabbit_db_user:clear_all_permissions_for_vhost(VirtualHost) of
+        {ok, Deletions} ->
+            lists:foreach(
+              fun (#topic_permission{topic_permission_key =
+                      #topic_permission_key{user_vhost =
+                         #user_vhost{username = Username}}}) ->
+                      rabbit_event:notify(
+                        topic_permission_deleted,
+                        [{user, Username},
+                         {vhost, VirtualHost},
+                         {user_who_performed_action, ActingUser}]);
+                  (#user_permission{user_vhost =
+                                    #user_vhost{username = Username}}) ->
+                      rabbit_event:notify(
+                        permission_deleted,
+                        [{user, Username},
+                         {vhost, VirtualHost},
+                         {user_who_performed_action, ActingUser}])
+              end, Deletions),
+            ok;
+        {error, _} = Err ->
+            Err
+    end.
 
 set_permissions_globally(Username, ConfigurePerm, WritePerm, ReadPerm, ActingUser) ->
     VirtualHosts = rabbit_vhost:list_names(),
@@ -642,9 +670,6 @@ clear_topic_permissions(Username, VirtualHost, Exchange, ActingUser) ->
             erlang:raise(Class, Error, Stacktrace)
     end.
 
-clear_topic_permissions_for_vhost(VirtualHost, _ActingUser) ->
-    rabbit_db_user:clear_matching_topic_permissions('_', VirtualHost, '_').
-
 put_user(User, ActingUser) -> put_user(User, undefined, ActingUser).
 
 put_user(User, Version, ActingUser) ->
diff --git a/deps/rabbit/src/rabbit_db_user.erl b/deps/rabbit/src/rabbit_db_user.erl
@@ -28,7 +28,8 @@
          set_topic_permissions/1,
          clear_topic_permissions/3,
          clear_matching_topic_permissions/3,
-         delete/1]).
+         delete/1,
+         clear_all_permissions_for_vhost/1]).
 
 -export([khepri_users_path/0,
          khepri_user_path/1,
@@ -548,6 +549,57 @@ clear_matching_user_permissions_in_khepri(Username, VHostName) ->
 any('_') -> ?KHEPRI_WILDCARD_STAR;
 any(Value) -> Value.
 
+%% -------------------------------------------------------------------
+%% clear_all_permissions_for_vhost().
+%% -------------------------------------------------------------------
+
+-spec clear_all_permissions_for_vhost(VHostName) -> Ret when
+      VHostName :: vhost:name(),
+      Ret :: {ok, DeletedPermissions} | {error, Reason :: any()},
+      DeletedPermissions :: [#topic_permission{} | #user_permission{}].
+%% @doc Transactionally deletes all user and topic permissions for a virtual
+%% host, returning any permissions that were deleted.
+%%
+%% @returns an OK-tuple with the deleted permissions or an error tuple if the
+%% operation could not be completed.
+%%
+%% @private
+
+clear_all_permissions_for_vhost(VHostName) when is_binary(VHostName) ->
+    rabbit_khepri:handle_fallback(
+      #{mnesia =>
+        fun() -> clear_all_permissions_for_vhost_in_mnesia(VHostName) end,
+        khepri =>
+        fun() -> clear_all_permissions_for_vhost_in_khepri(VHostName) end}).
+
+clear_all_permissions_for_vhost_in_mnesia(VHostName) ->
+    rabbit_mnesia:execute_mnesia_transaction(
+      fun() ->
+              Deletions =
+              clear_matching_topic_permissions_in_mnesia_tx(
+                '_', VHostName, '_') ++
+                  clear_matching_user_permissions_in_mnesia_tx(
+                    '_', VHostName),
+              {ok, Deletions}
+      end).
+
+clear_all_permissions_for_vhost_in_khepri(VHostName) ->
+    rabbit_khepri:transaction(
+      fun() ->
+              UserPermissionsPath = khepri_user_permission_path(
+                                      ?KHEPRI_WILDCARD_STAR, VHostName),
+              TopicPermissionsPath = khepri_topic_permission_path(
+                                       ?KHEPRI_WILDCARD_STAR, VHostName,
+                                       ?KHEPRI_WILDCARD_STAR),
+              {ok, UserProps} = khepri_tx_adv:delete_many(UserPermissionsPath),
+              {ok, TopicProps} = khepri_tx_adv:delete_many(
+                                   TopicPermissionsPath),
+              Deletions = rabbit_khepri:collect_payloads(
+                            TopicProps,
+                            rabbit_khepri:collect_payloads(UserProps)),
+              {ok, Deletions}
+      end, rw).
+
 %% -------------------------------------------------------------------
 %% get_topic_permissions().
 %% -------------------------------------------------------------------
diff --git a/deps/rabbit/src/rabbit_khepri.erl b/deps/rabbit/src/rabbit_khepri.erl
@@ -170,6 +170,10 @@
 
 -export([force_shrink_member_to_current_member/0]).
 
+%% Helpers for working with the Khepri API / types.
+-export([collect_payloads/1,
+         collect_payloads/2]).
+
 -ifdef(TEST).
 -export([force_metadata_store/1,
          clear_forced_metadata_store/0]).
@@ -983,6 +987,48 @@ info() ->
 handle_async_ret(RaEvent) ->
     khepri:handle_async_ret(?STORE_ID, RaEvent).
 
+%% -------------------------------------------------------------------
+%% collect_payloads().
+%% -------------------------------------------------------------------
+
+-spec collect_payloads(Props) -> Ret when
+      Props :: khepri:node_props(),
+      Ret :: [Payload],
+      Payload :: term().
+
+%% @doc Collects all payloads from a node props map.
+%%
+%% This is the same as calling `collect_payloads(Props, [])'.
+%%
+%% @private
+
+collect_payloads(Props) when is_map(Props) ->
+    collect_payloads(Props, []).
+
+-spec collect_payloads(Props, Acc0) -> Ret when
+      Props :: khepri:node_props(),
+      Acc0 :: [Payload],
+      Ret :: [Payload],
+      Payload :: term().
+
+%% @doc Collects all payloads from a node props map into the accumulator list.
+%%
+%% This is meant to be used with the `khepri_adv' API to easily collect the
+%% payloads from the return value of `khepri_adv:delete_many/4' for example.
+%%
+%% @returns all payloads in the node props map collected into a list, with
+%% `Acc0' as the tail.
+%%
+%% @private
+
+collect_payloads(Props, Acc0) when is_map(Props) andalso is_list(Acc0) ->
+    maps:fold(
+      fun (_Path, #{data := Payload}, Acc) ->
+              [Payload | Acc];
+          (_Path, _NoPayload, Acc) ->
+              Acc
+      end, Acc0, Props).
+
 %% -------------------------------------------------------------------
 %% if_has_data_wildcard().
 %% -------------------------------------------------------------------
diff --git a/deps/rabbit/src/rabbit_vhost.erl b/deps/rabbit/src/rabbit_vhost.erl
@@ -272,8 +272,7 @@ delete(VHost, ActingUser) ->
     %% calls would be responsible for the atomicity, not this code.
     %% Clear the permissions first to prohibit new incoming connections when deleting a vhost
     rabbit_log:info("Clearing permissions in vhost '~ts' because it's being deleted", [VHost]),
-    _ = rabbit_auth_backend_internal:clear_permissions_for_vhost(VHost, ActingUser),
-    _ = rabbit_auth_backend_internal:clear_topic_permissions_for_vhost(VHost, ActingUser),
+    ok = rabbit_auth_backend_internal:clear_all_permissions_for_vhost(VHost, ActingUser),
     rabbit_log:info("Deleting queues in vhost '~ts' because it's being deleted", [VHost]),
     QDelFun = fun (Q) -> rabbit_amqqueue:delete(Q, false, false, ActingUser) end,
     [begin
diff --git a/deps/rabbit/test/vhost_SUITE.erl b/deps/rabbit/test/vhost_SUITE.erl
@@ -33,6 +33,7 @@ groups() ->
         vhost_failure_forces_connection_closure,
         vhost_creation_idempotency,
         vhost_update_idempotency,
+        vhost_deletion,
         parse_tags
     ],
     ClusterSize2Tests = [
@@ -41,7 +42,8 @@ groups() ->
         vhost_failure_forces_connection_closure_on_failure_node,
         node_starts_with_dead_vhosts,
         node_starts_with_dead_vhosts_with_mirrors,
-        vhost_creation_idempotency
+        vhost_creation_idempotency,
+        vhost_deletion
     ],
     [
       {cluster_size_1_network, [], ClusterSize1Tests},
@@ -375,6 +377,118 @@ vhost_update_idempotency(Config) ->
         rabbit_ct_broker_helpers:delete_vhost(Config, VHost)
     end.
 
+vhost_deletion(Config) ->
+    VHost = <<"deletion-vhost">>,
+    ActingUser = <<"acting-user">>,
+    Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+
+    set_up_vhost(Config, VHost),
+
+    Conn = rabbit_ct_client_helpers:open_unmanaged_connection(Config, 0, VHost),
+    {ok, Chan} = amqp_connection:open_channel(Conn),
+
+    %% Declare some resources under the vhost. These should be deleted when the
+    %% vhost is deleted.
+    QName = <<"vhost-deletion-queue">>,
+    #'queue.declare_ok'{} = amqp_channel:call(
+                              Chan, #'queue.declare'{queue = QName, durable = true}),
+    XName = <<"vhost-deletion-exchange">>,
+    #'exchange.declare_ok'{} = amqp_channel:call(
+                                 Chan,
+                                 #'exchange.declare'{exchange = XName,
+                                                     durable = true,
+                                                     type = <<"direct">>}),
+    RoutingKey = QName,
+    #'queue.bind_ok'{} = amqp_channel:call(
+                           Chan,
+                           #'queue.bind'{exchange = XName,
+                                         queue = QName,
+                                         routing_key = RoutingKey}),
+    PolicyName = <<"ttl-policy">>,
+    rabbit_ct_broker_helpers:set_policy_in_vhost(
+      Config, Node, VHost,
+      PolicyName, <<"policy_ttl-queue">>, <<"all">>, [{<<"message-ttl">>, 20}],
+      ActingUser),
+
+    % Load the dummy event handler module on the node.
+    ok = rabbit_ct_broker_helpers:rpc(Config, Node, test_rabbit_event_handler, okay, []),
+    ok = rabbit_ct_broker_helpers:rpc(Config, Node, gen_event, add_handler,
+                                      [rabbit_event, test_rabbit_event_handler, []]),
+    try
+        rabbit_ct_broker_helpers:delete_vhost(Config, VHost),
+
+        Events0 = rabbit_ct_broker_helpers:rpc(Config, Node,
+                                               gen_event, call,
+                                               [rabbit_event, test_rabbit_event_handler, events, 1000]),
+        ct:pal(
+          ?LOW_IMPORTANCE,
+          "Events emitted during deletion: ~p", [lists:reverse(Events0)]),
+
+        %% Reorganize the event props into maps for easier matching.
+        Events = [{Type, maps:from_list(Props)} ||
+                  #event{type = Type, props = Props} <- Events0],
+
+        ?assertMatch(#{user := <<"guest">>, vhost := VHost},
+                     proplists:get_value(permission_deleted, Events)),
+
+        ?assertMatch(#{source_name := XName,
+                       source_kind := exchange,
+                       destination_name := QName,
+                       destination_kind := queue,
+                       routing_key := RoutingKey,
+                       vhost := VHost},
+                     proplists:get_value(binding_deleted, Events)),
+
+        ?assertMatch(#{name := #resource{name = QName,
+                                         kind = queue,
+                                         virtual_host = VHost}},
+                     proplists:get_value(queue_deleted, Events)),
+
+        ?assertEqual(
+          lists:sort([<<>>, <<"amq.direct">>, <<"amq.fanout">>, <<"amq.headers">>,
+                      <<"amq.match">>, <<"amq.rabbitmq.trace">>, <<"amq.topic">>,
+                      <<"vhost-deletion-exchange">>]),
+          lists:sort(lists:filtermap(
+                       fun ({exchange_deleted,
+                             #{name := #resource{name = Name}}}) ->
+                               {true, Name};
+                           (_Event) ->
+                               false
+                       end, Events))),
+
+        %% TODO: parameter_cleared and vhost_limits_cleared
+        %% ?assertMatch(#{name := PolicyName, vhost := VHost},
+        %%              proplists:get_value(policy_cleared, Events)),
+
+        ?assertMatch(#{name := VHost,
+                       user_who_performed_action := ActingUser},
+                     proplists:get_value(vhost_deleted, Events)),
+        ?assertMatch(#{name := VHost,
+                       node := Node,
+                       user_who_performed_action := ?INTERNAL_USER},
+                     proplists:get_value(vhost_down, Events)),
+
+        ?assert(proplists:is_defined(channel_closed, Events)),
+        ?assert(proplists:is_defined(connection_closed, Events)),
+
+        %% VHost deletion is not idempotent - we return an error - but deleting
+        %% the same vhost again should not cause any more resources to be
+        %% deleted. So we should see no new events in the `rabbit_event'
+        %% handler.
+        ?assertEqual(
+          {error, {no_such_vhost, VHost}},
+          rabbit_ct_broker_helpers:delete_vhost(Config, VHost)),
+        ?assertEqual(
+          Events0,
+          rabbit_ct_broker_helpers:rpc(
+            Config, Node,
+            gen_event, call,
+            [rabbit_event, test_rabbit_event_handler, events, 1000]))
+    after
+        rabbit_ct_broker_helpers:rpc(Config, Node,
+                                     gen_event, delete_handler, [rabbit_event, test_rabbit_event_handler, []])
+    end.
+
 vhost_is_created_with_default_limits(Config) ->
     VHost = <<"vhost1">>,
     Limits = [{<<"max-connections">>, 10}, {<<"max-queues">>, 1}],
