Add binding checks

Author: ansd

deps/rabbit/src/rabbit_access_control.erl

@@ -202,7 +202,7 @@ check_topic_access(User = #user{username = Username,
             check_access(
                 fun() -> Module:check_topic_access(
                     auth_user(User, Impl), Resource, Permission, Context) end,
-                Module, "~s access to topic '~ts' in exchange ~s refused for user '~ts'",
+                Module, "~s access to topic '~ts' in exchange ~ts refused for user '~ts'",
                 [Permission, maps:get(routing_key, Context), rabbit_misc:rs(Resource), Username]);
             (_, Else) -> Else
         end, ok, Modules).

deps/rabbit/src/rabbit_amqp_management.erl

@@ -24,7 +24,6 @@ handle_request(Request, Vhost, User, ConnectionPid) ->
     } = decode_req(ReqSections, {undefined, undefined}),
 
     {StatusCode,
-     RespAppProps0,
      RespBody} = try {PathSegments, QueryMap} = parse_uri(HttpRequestTarget),
                      handle_http_req(HttpMethod,
                                      PathSegments,
@@ -34,7 +33,9 @@ handle_request(Request, Vhost, User, ConnectionPid) ->
                                      User,
                                      ConnectionPid)
                  catch throw:{?MODULE, StatusCode0, Explanation} ->
-                           {StatusCode0, [], {utf8, unicode:characters_to_binary(Explanation)}}
+                           rabbit_log:warning("request ~ts ~ts failed: ~ts",
+                                              [HttpMethod, HttpRequestTarget, Explanation]),
+                           {StatusCode0, {utf8, Explanation}}
                  end,
 
     RespProps = #'v1_0.properties'{
@@ -44,7 +45,9 @@ handle_request(Request, Vhost, User, ConnectionPid) ->
                    %% [HTTP over AMQP WD 06 §5.1]
                    correlation_id = MessageId},
     RespAppProps = #'v1_0.application_properties'{
-                      content = [{{utf8, <<"http:response">>}, {utf8, <<"1.1">>}} | RespAppProps0]},
+                      content = [
+                                 {{utf8, <<"http:response">>}, {utf8, <<"1.1">>}}
+                                ]},
     RespDataSect = #'v1_0.amqp_value'{content = RespBody},
     RespSections = [RespProps, RespAppProps, RespDataSect],
     [amqp10_framing:encode_bin(Sect) || Sect <- RespSections].
@@ -75,7 +78,7 @@ handle_http_req(<<"PUT">>,
     Q0 = amqqueue:new(QName, none, Durable, AutoDelete, Owner,
                       QArgs, Vhost, #{user => Username}, QType),
     {new, _Q} = rabbit_queue_type:declare(Q0, node()),
-    {<<"201">>, [], null};
+    {<<"201">>, null};
 
 handle_http_req(<<"PUT">>,
                 [<<"exchanges">>, XNameBinQuoted],
@@ -95,8 +98,8 @@ handle_http_req(<<"PUT">>,
                 catch exit:#amqp_error{explanation = Explanation} ->
                           throw(<<"400">>, Explanation, [])
                 end,
-    ok = prohibit_default_exchange(XNameBin),
     XName = rabbit_misc:r(Vhost, exchange, XNameBin),
+    ok = prohibit_default_exchange(XName),
     ok = check_resource_access(XName, configure, User),
     X = case rabbit_exchange:lookup(XName) of
             {ok, FoundX} ->
@@ -111,7 +114,7 @@ handle_http_req(<<"PUT">>,
     try rabbit_exchange:assert_equivalence(
           X, XTypeAtom, Durable, AutoDelete, Internal, XArgs) of
         ok ->
-            {<<"204">>, [], null}
+            {<<"204">>, null}
     catch exit:#amqp_error{name = precondition_failed,
                            explanation = Expl} ->
               throw(<<"409">>, Expl, [])
@@ -132,7 +135,7 @@ handle_http_req(<<"DELETE">>,
                               rabbit_queue_type:purge(Q)
                       end),
     RespPayload = {map, [{{utf8, <<"message_count">>}, {ulong, NumMsgs}}]},
-    {<<"200">>, [], RespPayload};
+    {<<"200">>, RespPayload};
 
 handle_http_req(<<"DELETE">>,
                 [<<"queues">>, QNameBinQuoted],
@@ -145,7 +148,7 @@ handle_http_req(<<"DELETE">>,
     QName = rabbit_misc:r(Vhost, queue, QNameBin),
     {ok, NumMsgs} = rabbit_amqqueue:delete_with(QName, ConnPid, false, false, Username, true),
     RespPayload = {map, [{{utf8, <<"message_count">>}, {ulong, NumMsgs}}]},
-    {<<"200">>, [], RespPayload};
+    {<<"200">>, RespPayload};
 
 handle_http_req(<<"DELETE">>,
                 [<<"exchanges">>, XNameBinQuoted],
@@ -157,19 +160,19 @@ handle_http_req(<<"DELETE">>,
     XNameBin = uri_string:unquote(XNameBinQuoted),
     XName = rabbit_misc:r(Vhost, exchange, XNameBin),
     ok = prohibit_cr_lf(XNameBin),
-    ok = prohibit_default_exchange(XNameBin),
+    ok = prohibit_default_exchange(XName),
     ok = prohibit_reserved_amq(XName),
     ok = check_resource_access(XName, configure, User),
     _ = rabbit_exchange:delete(XName, false, Username),
-    {<<"204">>, [], null};
+    {<<"204">>, null};
 
 handle_http_req(<<"POST">>,
                 [<<"bindings">>],
                 _Query,
                 ReqPayload,
                 Vhost,
-                #user{username = Username},
-                _ConnPid) ->
+                User = #user{username = Username},
+                ConnPid) ->
     #{source := SrcXNameBin,
       binding_key := BindingKey,
       arguments := Args} = BindingMap = decode_binding(ReqPayload),
@@ -181,34 +184,37 @@ handle_http_req(<<"POST">>,
                             end,
     SrcXName = rabbit_misc:r(Vhost, exchange, SrcXNameBin),
     DstName = rabbit_misc:r(Vhost, DstKind, DstNameBin),
+    ok = binding_checks(SrcXName, DstName, BindingKey, User),
     Binding = #binding{source      = SrcXName,
                        destination = DstName,
                        key         = BindingKey,
                        args        = Args},
-    %%TODO If the binding already exists, return 303 with location.
-    ok = rabbit_binding:add(Binding, Username),
-    Location = compose_binding_uri(SrcXNameBin, DstKind, DstNameBin, BindingKey, Args),
-    AppProps = [{{utf8, <<"location">>}, {utf8, Location}}],
-    {<<"201">>, AppProps, null};
+    ok = binding_action(add, Binding, Username, ConnPid),
+    {<<"204">>, null};
 
 handle_http_req(<<"DELETE">>,
                 [<<"bindings">>, BindingSegment],
                 _Query,
                 null,
                 Vhost,
-                #user{username = Username},
-                _ConnPid) ->
-    {SrcXNameBin, DstKind, DstNameBin, BindingKey, ArgsHash} = decode_binding_path_segment(BindingSegment),
+                User = #user{username = Username},
+                ConnPid) ->
+    {SrcXNameBin,
+     DstKind,
+     DstNameBin,
+     BindingKey,
+     ArgsHash} = decode_binding_path_segment(BindingSegment),
     SrcXName = rabbit_misc:r(Vhost, exchange, SrcXNameBin),
     DstName = rabbit_misc:r(Vhost, DstKind, DstNameBin),
+    ok = binding_checks(SrcXName, DstName, BindingKey, User),
     Bindings = rabbit_binding:list_for_source_and_destination(SrcXName, DstName),
     case search_binding(BindingKey, ArgsHash, Bindings) of
         {value, Binding} ->
-            ok = rabbit_binding:remove(Binding, Username);
+            ok = binding_action(remove, Binding, Username, ConnPid);
         false ->
             ok
     end,
-    {<<"204">>, [], null};
+    {<<"204">>, null};
 
 handle_http_req(<<"GET">>,
                 [<<"bindings">>],
@@ -218,20 +224,23 @@ handle_http_req(<<"GET">>,
                 Vhost,
                 _User,
                 _ConnPid) ->
-    {DstKind, DstNameBin} = case QueryMap of
-                                #{<<"dste">> := DstX} ->
-                                    {exchange, DstX};
-                                #{<<"dstq">> := DstQ} ->
-                                    {queue, DstQ};
-                                _ ->
-                                    throw(<<"400">>, "missing 'dste' or 'dstq' in query: ~tp", QueryMap)
-                            end,
+    {DstKind,
+     DstNameBin} = case QueryMap of
+                       #{<<"dste">> := DstX} ->
+                           {exchange, DstX};
+                       #{<<"dstq">> := DstQ} ->
+                           {queue, DstQ};
+                       _ ->
+                           throw(<<"400">>,
+                                 "missing 'dste' or 'dstq' in query: ~tp",
+                                 QueryMap)
+                   end,
     SrcXName = rabbit_misc:r(Vhost, exchange, SrcXNameBin),
     DstName = rabbit_misc:r(Vhost, DstKind, DstNameBin),
     Bindings0 = rabbit_binding:list_for_source_and_destination(SrcXName, DstName),
     Bindings = [B || B = #binding{key = K} <- Bindings0, K =:= Key],
     RespPayload = encode_bindings(Bindings),
-    {<<"200">>, [], RespPayload}.
+    {<<"200">>, RespPayload}.
 
 decode_queue({map, KVList}) ->
     M = lists:foldl(
@@ -424,19 +433,47 @@ args_hash([]) ->
     <<>>;
 args_hash(Args)
   when is_list(Args) ->
+    %% Args is already sorted.
     Bin = <<(erlang:phash2(Args, 1 bsl 32)):32>>,
     base64:encode(Bin, #{mode => urlsafe,
                          padding => false}).
 
+-spec binding_checks(rabbit_types:exchange_name(),
+                     resource_name(),
+                     rabbit_types:binding_key(),
+                     rabbit_types:user()) -> ok.
+binding_checks(SrcXName, DstName, BindingKey, User) ->
+    lists:foreach(fun(#resource{name = NameBin} = Name) ->
+                          ok = prohibit_default_exchange(Name),
+                          ok = prohibit_cr_lf(NameBin)
+                  end, [SrcXName, DstName]),
+    ok = check_resource_access(DstName, write, User),
+    ok = check_resource_access(SrcXName, read, User),
+    case rabbit_exchange:lookup(SrcXName) of
+        {ok, SrcX} ->
+            rabbit_amqp_session:check_read_permitted_on_topic(SrcX, User, BindingKey);
+        {error, not_found} ->
+            ok
+    end.
+
+binding_action(Action, Binding, Username, ConnPid) ->
+    try rabbit_channel:binding_action(Action, Binding, Username, ConnPid)
+    catch exit:#amqp_error{explanation = Explanation} ->
+              throw(<<"400">>, Explanation, [])
+    end.
+
 prohibit_cr_lf(NameBin) ->
     case binary:match(NameBin, [<<"\n">>, <<"\r">>]) of
         nomatch ->
             ok;
         _Found ->
-            throw(<<"400">>, <<"Bad name '~ts': \n and \r not allowed">>, [NameBin])
+            throw(<<"400">>,
+                  <<"Bad name '~ts': line feed and carriage return characters not allowed">>,
+                  [NameBin])
     end.
 
-prohibit_default_exchange(<<>>) ->
+prohibit_default_exchange(#resource{kind = exchange,
+                                    name = <<"">>}) ->
     throw(<<"403">>, <<"operation not permitted on the default exchange">>, []);
 prohibit_default_exchange(_) ->
     ok.
@@ -464,6 +501,6 @@ check_resource_access(Resource, Perm, User) ->
 
 -spec throw(binary(), io:format(), [term()]) -> no_return().
 throw(StatusCode, Format, Data) ->
-    Explanation = lists:flatten(io_lib:format(Format, Data)),
-    rabbit_log:warning(Explanation),
-    throw({?MODULE, StatusCode, Explanation}).
+    Reason0 = lists:flatten(io_lib:format(Format, Data)),
+    Reason = unicode:characters_to_binary(Reason0),
+    throw({?MODULE, StatusCode, Reason}).

deps/rabbit/src/rabbit_amqp_session.erl

@@ -60,7 +60,9 @@
 -export([start_link/8,
          process_frame/2,
          list_local/0,
-         conserve_resources/3]).
+         conserve_resources/3,
+         check_read_permitted_on_topic/3
+        ]).
 
 -export([init/1,
          terminate/2,
@@ -2589,7 +2591,8 @@ check_topic_authorisation(#exchange{type = topic,
             try rabbit_access_control:check_topic_access(User, Resource, Permission, Context) of
                 ok ->
                     CacheTail = lists:sublist(Cache, ?MAX_PERMISSION_CACHE_SIZE - 1),
-                    put(?TOPIC_PERMISSION_CACHE, [CacheElem | CacheTail])
+                    put(?TOPIC_PERMISSION_CACHE, [CacheElem | CacheTail]),
+                    ok
             catch
                 exit:#amqp_error{name = access_refused,
                                  explanation = Msg} ->

deps/rabbit/src/rabbit_channel.erl

@@ -66,7 +66,8 @@
 -export([list_queue_states/1]).
 
 %% Mgmt HTTP API refactor
--export([handle_method/6]).
+-export([handle_method/6,
+         binding_action/4]).
 
 -import(rabbit_misc, [maps_put_truthy/3]).
 
@@ -1806,9 +1807,10 @@ queue_down_consumer_action(CTag, CMap) ->
         _            -> {recover, ConsumeSpec}
     end.
 
-binding_action(Fun, SourceNameBin0, DestinationType, DestinationNameBin0,
-               RoutingKey, Arguments, VHostPath, ConnPid, AuthzContext,
-               #user{username = Username} = User) ->
+binding_action_with_checks(
+  Action, SourceNameBin0, DestinationType, DestinationNameBin0,
+  RoutingKey, Arguments, VHostPath, ConnPid, AuthzContext,
+  #user{username = Username} = User) ->
     ExchangeNameBin = strip_cr_lf(SourceNameBin0),
     DestinationNameBin = strip_cr_lf(DestinationNameBin0),
     DestinationName = name_to_resource(DestinationType, DestinationNameBin, VHostPath),
@@ -1822,18 +1824,27 @@ binding_action(Fun, SourceNameBin0, DestinationType, DestinationNameBin0,
         {ok, Exchange}     ->
             check_read_permitted_on_topic(Exchange, User, RoutingKey, AuthzContext)
     end,
-    case Fun(#binding{source      = ExchangeName,
-                      destination = DestinationName,
-                      key         = RoutingKey,
-                      args        = Arguments},
-             fun (_X, Q) when ?is_amqqueue(Q) ->
-                     try rabbit_amqqueue:check_exclusive_access(Q, ConnPid)
-                     catch exit:Reason -> {error, Reason}
-                     end;
-                 (_X, #exchange{}) ->
-                     ok
-             end,
-             Username) of
+    Binding = #binding{source = ExchangeName,
+                       destination = DestinationName,
+                       key = RoutingKey,
+                       args = Arguments},
+    binding_action(Action, Binding, Username, ConnPid).
+
+-spec binding_action(add | remove,
+                     rabbit_types:binding(),
+                     rabbit_types:username(),
+                     pid()) -> ok.
+binding_action(Action, Binding, Username, ConnPid) ->
+    case rabbit_binding:Action(
+           Binding,
+           fun (_X, Q) when ?is_amqqueue(Q) ->
+                   try rabbit_amqqueue:check_exclusive_access(Q, ConnPid)
+                   catch exit:Reason -> {error, Reason}
+                   end;
+               (_X, #exchange{}) ->
+                   ok
+           end,
+           Username) of
         {error, {resources_missing, [{not_found, Name} | _]}} ->
             rabbit_amqqueue:not_found(Name);
         {error, {resources_missing, [{absent, Q, Reason} | _]}} ->
@@ -2362,33 +2373,33 @@ handle_method(#'exchange.bind'{destination = DestinationNameBin,
                                routing_key = RoutingKey,
                                arguments   = Arguments},
               ConnPid, AuthzContext, _CollectorId, VHostPath, User) ->
-    binding_action(fun rabbit_binding:add/3,
-                   SourceNameBin, exchange, DestinationNameBin,
-                   RoutingKey, Arguments, VHostPath, ConnPid, AuthzContext, User);
+    binding_action_with_checks(
+      add, SourceNameBin, exchange, DestinationNameBin,
+      RoutingKey, Arguments, VHostPath, ConnPid, AuthzContext, User);
 handle_method(#'exchange.unbind'{destination = DestinationNameBin,
                                  source      = SourceNameBin,
                                  routing_key = RoutingKey,
                                  arguments   = Arguments},
               ConnPid, AuthzContext, _CollectorId, VHostPath, User) ->
-    binding_action(fun rabbit_binding:remove/3,
-                       SourceNameBin, exchange, DestinationNameBin,
-                       RoutingKey, Arguments, VHostPath, ConnPid, AuthzContext, User);
+    binding_action_with_checks(
+      remove, SourceNameBin, exchange, DestinationNameBin,
+      RoutingKey, Arguments, VHostPath, ConnPid, AuthzContext, User);
 handle_method(#'queue.unbind'{queue       = QueueNameBin,
                               exchange    = ExchangeNameBin,
                               routing_key = RoutingKey,
                               arguments   = Arguments},
               ConnPid, AuthzContext, _CollectorId, VHostPath, User) ->
-    binding_action(fun rabbit_binding:remove/3,
-                   ExchangeNameBin, queue, QueueNameBin,
-                   RoutingKey, Arguments, VHostPath, ConnPid, AuthzContext, User);
+    binding_action_with_checks(
+      remove, ExchangeNameBin, queue, QueueNameBin,
+      RoutingKey, Arguments, VHostPath, ConnPid, AuthzContext, User);
 handle_method(#'queue.bind'{queue       = QueueNameBin,
                             exchange    = ExchangeNameBin,
                             routing_key = RoutingKey,
                             arguments   = Arguments},
               ConnPid, AuthzContext, _CollectorId, VHostPath, User) ->
-    binding_action(fun rabbit_binding:add/3,
-                   ExchangeNameBin, queue, QueueNameBin,
-                   RoutingKey, Arguments, VHostPath, ConnPid, AuthzContext, User);
+    binding_action_with_checks(
+      add, ExchangeNameBin, queue, QueueNameBin,
+      RoutingKey, Arguments, VHostPath, ConnPid, AuthzContext, User);
 %% Note that all declares to these are effectively passive. If it
 %% exists it by definition has one consumer.
 handle_method(#'queue.declare'{queue   = <<"amq.rabbitmq.reply-to",