Prevent malicious channels from sending messages to arbitrary processes with this mechanism.

2 files changed, 26 insertions, 3 deletions

diff --git a/src/pg_local.erl b/src/pg_local.erl
index f535b136..dc6401ca 100644
--- a/src/pg_local.erl
+++ b/src/pg_local.erl
@@ -34,7 +34,7 @@
 %%
 -module(pg_local).
 
--export([join/2, leave/2, get_members/1]).
+-export([join/2, leave/2, get_members/1, in_group/2]).
 -export([sync/0]). %% intended for testing only; not part of official API
 -export([start/0, start_link/0, init/1, handle_call/3, handle_cast/2,
          handle_info/2, terminate/2]).
@@ -50,6 +50,7 @@
 -spec(join/2 :: (name(), pid()) -> 'ok').
 -spec(leave/2 :: (name(), pid()) -> 'ok').
 -spec(get_members/1 :: (name()) -> [pid()]).
+-spec(in_group/2 :: (name(), pid()) -> boolean()).
 
 -spec(sync/0 :: () -> 'ok').
 
@@ -81,6 +82,10 @@ get_members(Name) ->
     ensure_started(),
     group_members(Name).
 
+in_group(Name, Pid) ->
+    ensure_started(),
+    member_present(Name, Pid).
+
 sync() ->
     ensure_started(),
     gen_server:call(?MODULE, sync, infinity).
@@ -199,6 +204,12 @@ member_in_group(Pid, Name) ->
     [{{member, Name, Pid}, N}] = ets:lookup(pg_local_table, {member, Name, Pid}),
     lists:duplicate(N, Pid).
 
+member_present(Name, Pid) ->
+    case ets:lookup(pg_local_table, {member, Name, Pid}) of
+        [_] -> true;
+        []  -> false
+    end.
+
 member_groups(Pid) ->
     [Name || [Name] <- ets:match(pg_local_table, {{pid, Pid, '$1'}})].
 
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index a153822e..b61a53f8 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -31,7 +31,7 @@
          handle_info/2, handle_pre_hibernate/1, prioritise_call/4,
          prioritise_cast/3, prioritise_info/3, format_message_queue/2]).
 %% Internal
--export([list_local/0]).
+-export([list_local/0, deliver_reply_local/3]).
 
 -record(ch, {state, protocol, channel, reader_pid, writer_pid, conn_pid,
              conn_name, limiter, tx, next_tag, unacked_message_q, user,
@@ -97,6 +97,9 @@
 -spec(deliver/4 ::
         (pid(), rabbit_types:ctag(), boolean(), rabbit_amqqueue:qmsg())
         -> 'ok').
+-spec(deliver_reply/2 :: (binary(), rabbit_types:delivery()) -> 'ok').
+-spec(deliver_reply_local/3 ::
+        (pid(), binary(), rabbit_types:delivery()) -> 'ok').
 -spec(send_credit_reply/2 :: (pid(), non_neg_integer()) -> 'ok').
 -spec(send_drained/2 :: (pid(), [{rabbit_types:ctag(), non_neg_integer()}])
                         -> 'ok').
@@ -146,7 +149,16 @@ deliver(Pid, ConsumerTag, AckRequired, Msg) ->
 deliver_reply(<<"amq.rabbitmq.reply-to.", Rest/binary>>, Delivery) ->
     [PidEnc, Key] = binary:split(Rest, <<".">>),
     Pid = binary_to_term(base64:decode(PidEnc)),
-    gen_server2:cast(Pid, {deliver_reply, Key, Delivery}).
+    delegate:invoke_no_result(
+      Pid, {?MODULE, deliver_reply_local, [Key, Delivery]}).
+
+%% We want to ensure people can't use this mechanism to send a message
+%% to an arbitrary process and kill it!
+deliver_reply_local(Pid, Key, Delivery) ->
+    case pg_local:in_group(rabbit_channels, Pid) of
+        true  -> gen_server2:cast(Pid, {deliver_reply, Key, Delivery});
+        false -> ok
+    end.
 
 send_credit_reply(Pid, Len) ->
     gen_server2:cast(Pid, {send_credit_reply, Len}).