Merge pull request #2050 from rabbitmq/oauth2-credential-expiration-support

Support client-driven refresh of expiring credentials

5 files changed, 106 insertions, 11 deletions

diff --git a/src/rabbit_access_control.erl b/src/rabbit_access_control.erl
index d04f0047de..4c68fe2eab 100644
--- a/src/rabbit_access_control.erl
+++ b/src/rabbit_access_control.erl
@@ -21,6 +21,8 @@
 -export([check_user_pass_login/2, check_user_login/2, check_user_loopback/2,
          check_vhost_access/4, check_resource_access/4, check_topic_access/4]).
 
+-export([permission_cache_can_expire/1, update_state/2]).
+
 %%----------------------------------------------------------------------------
 
 -export_type([permission_atom/0]).
@@ -217,3 +219,38 @@ check_access(Fun, Module, ErrStr, ErrArgs, ErrName) ->
             rabbit_log:error(FullErrStr, FullErrArgs),
             rabbit_misc:protocol_error(ErrName, FullErrStr, FullErrArgs)
     end.
+
+-spec update_state(User :: rabbit_types:user(), NewState :: term()) ->
+    {'ok', rabbit_types:auth_user()} |
+    {'refused', string()} |
+    {'error', any()}.
+
+update_state(User = #user{authz_backends = Backends0}, NewState) ->
+    %% N.B.: we use foldl/3 and prepending, so the final list of
+    %% backends is in reverse order from the original list.
+    Backends = lists:foldl(
+                fun({Module, Impl}, {ok, Acc}) ->
+                        case Module:state_can_expire() of
+                          true  ->
+                            case Module:update_state(auth_user(User, Impl), NewState) of
+                              {ok, #auth_user{impl = Impl1}} ->
+                                {ok, [{Module, Impl1} | Acc]};
+                              Else -> Else
+                            end;
+                          false ->
+                            {ok, [{Module, Impl} | Acc]}
+                        end;
+                   (_, {error, _} = Err)      -> Err;
+                   (_, {refused, _, _} = Err) -> Err
+                end, {ok, []}, Backends0),
+    case Backends of
+      {ok, Pairs} -> {ok, User#user{authz_backends = lists:reverse(Pairs)}};
+      Else        -> Else
+    end.
+
+-spec permission_cache_can_expire(User :: rabbit_types:user()) -> boolean().
+
+%% Returns true if any of the backends support credential expiration,
+%% otherwise returns false.
+permission_cache_can_expire(#user{authz_backends = Backends}) ->
+    lists:any(fun ({Module, _State}) -> Module:state_can_expire() end, Backends).
diff --git a/src/rabbit_auth_backend_internal.erl b/src/rabbit_auth_backend_internal.erl
index e16b14734b..e675ad188b 100644
--- a/src/rabbit_auth_backend_internal.erl
+++ b/src/rabbit_auth_backend_internal.erl
@@ -40,6 +40,8 @@
          list_user_vhost_permissions/2,
          list_user_topic_permissions/1, list_vhost_topic_permissions/1, list_user_vhost_topic_permissions/2]).
 
+-export([state_can_expire/0]).
+
 %% for testing
 -export([hashing_module_for_user/1, expand_topic_permission/2]).
 
@@ -93,6 +95,8 @@ user_login_authentication(Username, AuthProps) ->
         false -> exit({unknown_auth_props, Username, AuthProps})
     end.
 
+state_can_expire() -> false.
+
 user_login_authorization(Username, _AuthProps) ->
     case user_login_authentication(Username, []) of
         {ok, #auth_user{impl = Impl, tags = Tags}} -> {ok, Impl, Tags};
@@ -123,7 +127,7 @@ check_vhost_access(#auth_user{username = Username}, VHostPath, _AuthzData) ->
 
 check_resource_access(#auth_user{username = Username},
                       #resource{virtual_host = VHostPath, name = Name},
-                      Permission, 
+                      Permission,
                       _AuthContext) ->
     case mnesia:dirty_read({rabbit_user_permission,
                             #user_vhost{username     = Username,
diff --git a/src/rabbit_auth_mechanism_plain.erl b/src/rabbit_auth_mechanism_plain.erl
index cfc1a0ce18..706e5eedfa 100644
--- a/src/rabbit_auth_mechanism_plain.erl
+++ b/src/rabbit_auth_mechanism_plain.erl
@@ -31,9 +31,6 @@
 %% SASL PLAIN, as used by the Qpid Java client and our clients. Also,
 %% apparently, by OpenAMQ.
 
-%% TODO: reimplement this using the binary module? - that makes use of
-%% BIFs to do binary matching and will thus be much faster.
-
 description() ->
     [{description, <<"SASL PLAIN authentication mechanism">>}].
 
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index d4c8ba4056..d16929d962 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,
@@ -460,11 +460,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:auth_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.
 
@@ -490,6 +500,8 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
                   _ ->
                       Limiter0
               end,
+    %% Process dictionary is used here because permission cache already uses it. MK.
+    put(permission_cache_can_expire, rabbit_access_control:permission_cache_can_expire(User)),
     MaxMessageSize = get_max_message_size(),
     ConsumerTimeout = get_consumer_timeout(),
     State = #ch{cfg = #conf{state = starting,
@@ -842,6 +854,10 @@ handle_info({{Ref, Node}, LateAnswer},
     noreply(State);
 
 handle_info(tick, State0 = #ch{queue_states = QueueStates0}) ->
+    case get(permission_cache_can_expire) of
+      true  -> ok = clear_permission_cache();
+      _     -> ok
+    end,
     QueueStates1 =
         maps:filter(fun(_, QS) ->
                             QName = rabbit_quorum_queue:queue_name(QS),
@@ -854,7 +870,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(),
@@ -977,7 +996,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
@@ -1055,8 +1074,8 @@ check_topic_authorisation(_, _, _, _, _, _) ->
     ok.
 
 check_topic_authorisation(#exchange{name = Name = #resource{virtual_host = VHost}, type = topic},
-                          User = #user{username = Username},
-                          AmqpParams, RoutingKey, Permission) ->
+                             User = #user{username = Username},
+                             AmqpParams, RoutingKey, Permission) ->
     Resource = Name#resource{kind = topic},
     VariableMap = build_topic_variable_map(AmqpParams, VHost, Username),
     Context = #{routing_key  => RoutingKey,
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 6f3e993beb..39ac0ef8ac 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -1274,6 +1274,44 @@ handle_method0(#'connection.close_ok'{},
                State = #v1{connection_state = closed}) ->
     self() ! terminate_connection,
     State;
+handle_method0(#'connection.update_secret'{new_secret = NewSecret, reason = Reason},
+               State = #v1{connection =
+                               #connection{protocol   = Protocol,
+                                           user       = User = #user{username = Username},
+                                           log_name   = ConnName} = Conn,
+                           sock       = Sock}) when ?IS_RUNNING(State) ->
+    rabbit_log_connection:debug(
+        "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} ->
+        %% 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): "
+            "user '~s' updated secret, reason: ~s~n",
+            [self(), dynamic_connection_name(ConnName), Username, Reason]),
+        State#v1{connection = Conn#connection{user = User1}};
+      {refused, Message} ->
+        rabbit_log_connection:error("Secret update was refused for user '~p': ~p",
+                                    [Username, Message]),
+        rabbit_misc:protocol_error(not_allowed, "New secret was refused by one of the backends", []);
+      {error, Message} ->
+        rabbit_log_connection:error("Secret update for user '~p' failed: ~p",
+                                    [Username, Message]),
+        rabbit_misc:protocol_error(not_allowed,
+                                  "Secret update failed", [])
+    end;
 handle_method0(_Method, State) when ?IS_STOPPING(State) ->
     State;
 handle_method0(_Method, #v1{connection_state = S}) ->