Propagate updated user/authn/authz state to channels

Channel processes use it for authz backend checks. The record
is identical to that of connection process (and originates from it), so
a simple cast to update the field is appropriate: any errors would
be handled and communicated back by the connection itself.

Per discussion with @acogoluegnes and @kjnilsson.

2 files changed, 31 insertions, 8 deletions

diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index f5c9e8dfce..852b03e859 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -63,7 +63,7 @@
 -export([refresh_config_local/0, ready_for_close/1]).
 -export([refresh_interceptors/0]).
 -export([force_event_refresh/1]).
--export([source/2]).
+-export([source/2, update_user_state/2]).
 
 -export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2,
          handle_info/2, handle_pre_hibernate/1, handle_post_hibernate/1,
@@ -459,11 +459,21 @@ force_event_refresh(Ref) ->
 list_queue_states(Pid) ->
     gen_server2:call(Pid, list_queue_states).
 
--spec source(pid(), any()) -> any().
+-spec source(pid(), any()) -> 'ok' | {error, channel_terminated}.
 
 source(Pid, Source) when is_pid(Pid) ->
     case erlang:is_process_alive(Pid) of
-        true  -> Pid ! {channel_source, Source};
+        true  -> Pid ! {channel_source, Source},
+                 ok;
+        false -> {error, channel_terminated}
+    end.
+
+-spec update_user_state(pid(), rabbit_types:user()) -> 'ok' | {error, channel_terminated}.
+
+update_user_state(Pid, UserState) when is_pid(Pid) ->
+    case erlang:is_process_alive(Pid) of
+        true  -> Pid ! {update_user_state, UserState},
+                 ok;
         false -> {error, channel_terminated}
     end.
 
@@ -850,7 +860,10 @@ handle_info(tick, State0 = #ch{queue_states = QueueStates0}) ->
             Return
     end;
 handle_info({channel_source, Source}, State = #ch{cfg = Cfg}) ->
-    noreply(State#ch{cfg = Cfg#conf{source = Source}}).
+    noreply(State#ch{cfg = Cfg#conf{source = Source}});
+handle_info({update_user_state, User}, State = #ch{cfg = Cfg}) ->
+    noreply(State#ch{cfg = Cfg#conf{user = User}}).
+
 
 handle_pre_hibernate(State0) ->
     ok = clear_permission_cache(),
@@ -973,7 +986,7 @@ return_queue_declare_ok(#resource{name = ActualName},
 
 check_resource_access(User, Resource, Perm, Context) ->
     V = {Resource, Context, Perm},
-    
+
     Cache = case get(permission_cache) of
                 undefined -> [];
                 Other     -> Other
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 5eb8032bea..741c91b082 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -1280,12 +1280,22 @@ handle_method0(#'connection.update_secret'{new_secret = NewSecret, reason = Reas
                                            log_name   = ConnName},
                            sock       = Sock}) when ?IS_RUNNING(State) ->
     rabbit_log_connection:debug(
-        "connection ~p (~s): "
-        "user '~s' attempts to update secret, reason: ~s~n",
+        "connection ~p (~s) of user '~s': "
+        "asked to update secret, reason: ~s~n",
         [self(), dynamic_connection_name(ConnName), Username, Reason]),
     case rabbit_access_control:update_state(User, NewSecret) of
       {ok, User1} ->
-        rabbit_log_connection:debug("User1: ~p", [User1]),
+        rabbit_log_connection:debug("Updated user/auth state: ~p", [User1]),
+        %% User/auth backend state has been updated. Now we can propagate it to channels
+        %% asynchronously and return. All the channels have to do is to update their
+        %% own state.
+        %%
+        %% Any secret update errors coming from the authz backend will be handled in the other branch.
+        %% Therefore we optimistically do no error handling here. MK.
+        lists:foreach(fun(Ch) ->
+          rabbit_log:debug("Updating user/auth backend state for channel ~p", [Ch]),
+          rabbit_channel:update_user_state(Ch, User1)
+        end, all_channels()),
         ok = send_on_channel0(Sock, #'connection.update_secret_ok'{}, Protocol),
         rabbit_log_connection:info(
             "connection ~p (~s): "