Propagate protocol-specific context to authorization

Propagate protocol-specific context like MQTT client ID to
authentication and authorization backends. The propagation was already
there for topic authorization. Resource check needed propagation and a
new argument in the backend function. Virtual host check just needed
propagation, as there was already a context parameter (for peer
address).

A context parameter and propagation was already there for login-related
functions.

Unfortunately the context API is not fully consistent, due to historical
reasons: proplists are the historical data structure, maps are used
whenever possible.

References #1767

5 files changed, 69 insertions, 45 deletions

diff --git a/src/rabbit_access_control.erl b/src/rabbit_access_control.erl
index 954d003991..d04f0047de 100644
--- a/src/rabbit_access_control.erl
+++ b/src/rabbit_access_control.erl
@@ -19,7 +19,7 @@
 -include("rabbit.hrl").
 
 -export([check_user_pass_login/2, check_user_login/2, check_user_loopback/2,
-         check_vhost_access/3, check_resource_access/3, check_topic_access/4]).
+         check_vhost_access/4, check_resource_access/4, check_topic_access/4]).
 
 %%----------------------------------------------------------------------------
 
@@ -43,6 +43,7 @@ check_user_pass_login(Username, Password) ->
             {'refused', rabbit_types:username(), string(), [any()]}.
 
 check_user_login(Username, AuthProps) ->
+    %% extra auth properties like MQTT client id are in AuthProps
     {ok, Modules} = application:get_env(rabbit, auth_backends),
     R = lists:foldl(
           fun ({ModN, ModZs0}, {refused, _, _, _}) ->
@@ -137,18 +138,20 @@ get_authz_data_from(undefined) ->
 % is enabled and it's a direct connection.
 -spec check_vhost_access(User :: rabbit_types:user(),
                          VHostPath :: rabbit_types:vhost(),
-                         AuthzRawData :: {socket, rabbit_net:socket()} | {ip, inet:ip_address() | binary()} | undefined) ->
+                         AuthzRawData :: {socket, rabbit_net:socket()} | {ip, inet:ip_address() | binary()} | undefined,
+                         AuthzContext :: map()) ->
     'ok' | rabbit_types:channel_exit().
 check_vhost_access(User = #user{username       = Username,
-                                authz_backends = Modules}, VHostPath, AuthzRawData) ->
+                                authz_backends = Modules}, VHostPath, AuthzRawData, AuthzContext) ->
     AuthzData = get_authz_data_from(AuthzRawData),
+    FullAuthzContext = create_vhost_access_authz_data(AuthzData, AuthzContext),
     lists:foldl(
       fun({Mod, Impl}, ok) ->
               check_access(
                 fun() ->
                         rabbit_vhost:exists(VHostPath) andalso
                             Mod:check_vhost_access(
-                              auth_user(User, Impl), VHostPath, AuthzData)
+                              auth_user(User, Impl), VHostPath, FullAuthzContext)
                 end,
                 Mod, "access to vhost '~s' refused for user '~s'",
                 [VHostPath, Username], not_allowed);
@@ -156,22 +159,31 @@ check_vhost_access(User = #user{username       = Username,
               Else
       end, ok, Modules).
 
+create_vhost_access_authz_data(undefined, Context) when map_size(Context) == 0 ->
+    undefined;
+create_vhost_access_authz_data(undefined, Context) ->
+    Context;
+create_vhost_access_authz_data(PeerAddr, Context) when map_size(Context) == 0 ->
+    PeerAddr;
+create_vhost_access_authz_data(PeerAddr, Context) ->
+    maps:merge(PeerAddr, Context).
+
 -spec check_resource_access
-        (rabbit_types:user(), rabbit_types:r(atom()), permission_atom()) ->
+        (rabbit_types:user(), rabbit_types:r(atom()), permission_atom(), rabbit_types:authz_context()) ->
             'ok' | rabbit_types:channel_exit().
 
 check_resource_access(User, R = #resource{kind = exchange, name = <<"">>},
-                      Permission) ->
+                      Permission, Context) ->
     check_resource_access(User, R#resource{name = <<"amq.default">>},
-                          Permission);
+                          Permission, Context);
 check_resource_access(User = #user{username       = Username,
                                    authz_backends = Modules},
-                      Resource, Permission) ->
+                      Resource, Permission, Context) ->
     lists:foldl(
       fun({Module, Impl}, ok) ->
               check_access(
                 fun() -> Module:check_resource_access(
-                           auth_user(User, Impl), Resource, Permission) end,
+                           auth_user(User, Impl), Resource, Permission, Context) end,
                 Module, "access to ~s refused for user '~s'",
                 [rabbit_misc:rs(Resource), Username]);
          (_, Else) -> Else
diff --git a/src/rabbit_auth_backend_internal.erl b/src/rabbit_auth_backend_internal.erl
index 2f8e85f0f3..e16b14734b 100644
--- a/src/rabbit_auth_backend_internal.erl
+++ b/src/rabbit_auth_backend_internal.erl
@@ -21,7 +21,7 @@
 -behaviour(rabbit_authz_backend).
 
 -export([user_login_authentication/2, user_login_authorization/2,
-         check_vhost_access/3, check_resource_access/3, check_topic_access/4]).
+         check_vhost_access/3, check_resource_access/4, check_topic_access/4]).
 
 -export([add_user/3, delete_user/2, lookup_user/1,
          change_password/3, clear_password/2,
@@ -123,7 +123,8 @@ 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,
                                         virtual_host = VHostPath}}) of
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 7c07aa07f3..95691f25cd 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -970,8 +970,9 @@ return_queue_declare_ok(#resource{name = ActualName},
                                           message_count  = MessageCount,
                                           consumer_count = ConsumerCount}).
 
-check_resource_access(User, Resource, Perm) ->
-    V = {Resource, Perm},
+check_resource_access(User, Resource, Perm, Context) ->
+    V = {Resource, Context, Perm},
+    
     Cache = case get(permission_cache) of
                 undefined -> [];
                 Other     -> Other
@@ -979,7 +980,7 @@ check_resource_access(User, Resource, Perm) ->
     case lists:member(V, Cache) of
         true  -> ok;
         false -> ok = rabbit_access_control:check_resource_access(
-                        User, Resource, Perm),
+                        User, Resource, Perm, Context),
                  CacheTail = lists:sublist(Cache, ?MAX_PERMISSION_CACHE_SIZE-1),
                  put(permission_cache, [V | CacheTail])
     end.
@@ -988,14 +989,14 @@ clear_permission_cache() -> erase(permission_cache),
                             erase(topic_permission_cache),
                             ok.
 
-check_configure_permitted(Resource, User) ->
-    check_resource_access(User, Resource, configure).
+check_configure_permitted(Resource, User, Context) ->
+    check_resource_access(User, Resource, configure, Context).
 
-check_write_permitted(Resource, User) ->
-    check_resource_access(User, Resource, write).
+check_write_permitted(Resource, User, Context) ->
+    check_resource_access(User, Resource, write, Context).
 
-check_read_permitted(Resource, User) ->
-    check_resource_access(User, Resource, read).
+check_read_permitted(Resource, User, Context) ->
+    check_resource_access(User, Resource, read, Context).
 
 check_write_permitted_on_topic(Resource, User, ConnPid, RoutingKey, ChSrc) ->
     check_topic_authorisation(Resource, User, ConnPid, RoutingKey, ChSrc, write).
@@ -1081,16 +1082,19 @@ get_amqp_params(ConnPid, true, Timeout) ->
     rabbit_amqp_connection:amqp_params(ConnPid, Timeout).
 
 build_topic_variable_map(AmqpParams, VHost, Username) ->
-    VariableFromAmqpParams = extract_topic_variable_map_from_amqp_params(AmqpParams),
+    VariableFromAmqpParams = extract_variable_map_from_amqp_params(AmqpParams),
     maps:merge(VariableFromAmqpParams, #{<<"vhost">> => VHost, <<"username">> => Username}).
 
+extract_authz_context(ConnPid, ChSrc) ->
+    extract_variable_map_from_amqp_params(get_amqp_params(ConnPid, ChSrc)).
+
 %% use tuple representation of amqp_params to avoid coupling.
 %% get variable map only from amqp_params_direct, not amqp_params_network.
 %% amqp_params_direct are usually used from plugins (e.g. MQTT, STOMP)
-extract_topic_variable_map_from_amqp_params([{amqp_params, {amqp_params_direct, _, _, _, _,
+extract_variable_map_from_amqp_params([{amqp_params, {amqp_params_direct, _, _, _, _,
                                              {amqp_adapter_info, _,_,_,_,_,_,AdditionalInfo}, _}}]) ->
     proplists:get_value(variable_map, AdditionalInfo, #{});
-extract_topic_variable_map_from_amqp_params(_) ->
+extract_variable_map_from_amqp_params(_) ->
     #{}.
 
 check_msg_size(Content, MaxMessageSize) ->
@@ -1297,7 +1301,7 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
                                    }) ->
     check_msg_size(Content, MaxMessageSize),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
-    check_write_permitted(ExchangeName, User),
+    check_write_permitted(ExchangeName, User, extract_authz_context(ConnPid, ChSrc)),
     Exchange = rabbit_exchange:lookup_or_die(ExchangeName),
     check_internal_exchange(Exchange),
     check_write_permitted_on_topic(Exchange, User, ConnPid, RoutingKey, ChSrc),
@@ -1352,13 +1356,14 @@ handle_method(#'basic.get'{queue = QueueNameBin, no_ack = NoAck},
               _, State = #ch{cfg = #conf{writer_pid = WriterPid,
                                          conn_pid = ConnPid,
                                          user = User,
-                                         virtual_host = VHostPath
+                                         virtual_host = VHostPath,
+                                         source = ChSrc
                                         },
                              limiter = Limiter,
                              next_tag   = DeliveryTag,
                              queue_states = QueueStates0}) ->
     QueueName = qbin_to_resource(QueueNameBin, VHostPath),
-    check_read_permitted(QueueName, User),
+    check_read_permitted(QueueName, User, extract_authz_context(ConnPid, ChSrc)),
     case rabbit_amqqueue:with_exclusive_access_or_die(
            QueueName, ConnPid,
            %% Use the delivery tag as consumer tag for quorum queues
@@ -1437,13 +1442,15 @@ handle_method(#'basic.consume'{queue        = QueueNameBin,
                                arguments    = Args},
               _, State = #ch{cfg = #conf{consumer_prefetch = ConsumerPrefetch,
                                          user = User,
-                                         virtual_host = VHostPath},
+                                         virtual_host = VHostPath,
+                                         conn_pid = ConnPid,
+                                         source = ChSrc},
                              consumer_mapping  = ConsumerMapping
                             }) ->
     case maps:find(ConsumerTag, ConsumerMapping) of
         error ->
             QueueName = qbin_to_resource(QueueNameBin, VHostPath),
-            check_read_permitted(QueueName, User),
+            check_read_permitted(QueueName, User, extract_authz_context(ConnPid, ChSrc)),
             ActualConsumerTag =
                 case ConsumerTag of
                     <<>>  -> rabbit_guid:binary(rabbit_guid:gen_secure(),
@@ -1915,10 +1922,11 @@ binding_action(Fun, SourceNameBin0, DestinationType, DestinationNameBin0,
     ExchangeNameBin = strip_cr_lf(SourceNameBin0),
     DestinationNameBin = strip_cr_lf(DestinationNameBin0),
     DestinationName = name_to_resource(DestinationType, DestinationNameBin, VHostPath),
-    check_write_permitted(DestinationName, User),
+    AuthContext = extract_authz_context(ConnPid, ChSrc),
+    check_write_permitted(DestinationName, User, AuthContext),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
     [check_not_default_exchange(N) || N <- [DestinationName, ExchangeName]],
-    check_read_permitted(ExchangeName, User),
+    check_read_permitted(ExchangeName, User, AuthContext),
     case rabbit_exchange:lookup(ExchangeName) of
         {error, not_found} ->
             ok;
@@ -2486,7 +2494,7 @@ handle_method(#'queue.declare'{queue       = QueueNameBin,
                         Other -> check_name('queue', Other)
                     end,
     QueueName = rabbit_misc:r(VHostPath, queue, ActualNameBin),
-    check_configure_permitted(QueueName, User),
+    check_configure_permitted(QueueName, User, extract_authz_context(ConnPid, ChSrc)),
     rabbit_core_metrics:queue_declared(QueueName),
     case rabbit_amqqueue:with(
            QueueName,
@@ -2508,8 +2516,9 @@ handle_method(#'queue.declare'{queue       = QueueNameBin,
                       "invalid type '~s' for arg '~s' in ~s",
                       [Type, DlxKey, rabbit_misc:rs(QueueName)]);
                DLX ->
-                   check_read_permitted(QueueName, User),
-                   check_write_permitted(DLX, User),
+                   AuthContext = extract_authz_context(ConnPid, ChSrc),
+                   check_read_permitted(QueueName, User, AuthContext),
+                   check_write_permitted(DLX, User, AuthContext),
                    ok
             end,
             case rabbit_amqqueue:declare(QueueName, Durable, AutoDelete,
@@ -2561,12 +2570,12 @@ handle_method(#'queue.declare'{queue   = QueueNameBin,
 handle_method(#'queue.delete'{queue     = QueueNameBin,
                               if_unused = IfUnused,
                               if_empty  = IfEmpty},
-              ConnPid, _ChSrc, _CollectorPid, VHostPath,
+              ConnPid, ChSrc, _CollectorPid, VHostPath,
               User = #user{username = Username}) ->
     StrippedQueueNameBin = strip_cr_lf(QueueNameBin),
     QueueName = qbin_to_resource(StrippedQueueNameBin, VHostPath),
 
-    check_configure_permitted(QueueName, User),
+    check_configure_permitted(QueueName, User, extract_authz_context(ConnPid, ChSrc)),
     case rabbit_amqqueue:with(
            QueueName,
            fun (Q) ->
@@ -2590,13 +2599,13 @@ handle_method(#'queue.delete'{queue     = QueueNameBin,
     end;
 handle_method(#'exchange.delete'{exchange  = ExchangeNameBin,
                                  if_unused = IfUnused},
-              _ConnPid, _ChSrc, _CollectorPid, VHostPath,
+              ConnPid, ChSrc, _CollectorPid, VHostPath,
               User = #user{username = Username}) ->
     StrippedExchangeNameBin = strip_cr_lf(ExchangeNameBin),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, StrippedExchangeNameBin),
     check_not_default_exchange(ExchangeName),
     check_exchange_deletion(ExchangeName),
-    check_configure_permitted(ExchangeName, User),
+    check_configure_permitted(ExchangeName, User, extract_authz_context(ConnPid, ChSrc)),
     case rabbit_exchange:delete(ExchangeName, IfUnused, Username) of
         {error, not_found} ->
             ok;
@@ -2606,9 +2615,9 @@ handle_method(#'exchange.delete'{exchange  = ExchangeNameBin,
             ok
     end;
 handle_method(#'queue.purge'{queue = QueueNameBin},
-              ConnPid, _ChSrc, _CollectorPid, VHostPath, User) ->
+              ConnPid, ChSrc, _CollectorPid, VHostPath, User) ->
     QueueName = qbin_to_resource(QueueNameBin, VHostPath),
-    check_read_permitted(QueueName, User),
+    check_read_permitted(QueueName, User, extract_authz_context(ConnPid, ChSrc)),
     rabbit_amqqueue:with_exclusive_access_or_die(
       QueueName, ConnPid,
       fun (Q) -> rabbit_amqqueue:purge(Q) end);
@@ -2619,12 +2628,12 @@ handle_method(#'exchange.declare'{exchange    = ExchangeNameBin,
                                   auto_delete = AutoDelete,
                                   internal    = Internal,
                                   arguments   = Args},
-              _ConnPid, _ChSrc, _CollectorPid, VHostPath,
+              ConnPid, ChSrc, _CollectorPid, VHostPath,
               #user{username = Username} = User) ->
     CheckedType = rabbit_exchange:check_type(TypeNameBin),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, strip_cr_lf(ExchangeNameBin)),
     check_not_default_exchange(ExchangeName),
-    check_configure_permitted(ExchangeName, User),
+    check_configure_permitted(ExchangeName, User, extract_authz_context(ConnPid, ChSrc)),
     X = case rabbit_exchange:lookup(ExchangeName) of
             {ok, FoundX} -> FoundX;
             {error, not_found} ->
@@ -2636,8 +2645,9 @@ handle_method(#'exchange.declare'{exchange    = ExchangeNameBin,
                         precondition_failed(
                           "invalid type '~s' for arg '~s' in ~s",
                           [Type, AeKey, rabbit_misc:rs(ExchangeName)]);
-                    AName     -> check_read_permitted(ExchangeName, User),
-                                 check_write_permitted(AName, User),
+                    AName     -> AuthContext = extract_authz_context(ConnPid, ChSrc),
+                                 check_read_permitted(ExchangeName, User, AuthContext),
+                                 check_write_permitted(AName, User, AuthContext),
                                  ok
                 end,
                 rabbit_exchange:declare(ExchangeName,
diff --git a/src/rabbit_direct.erl b/src/rabbit_direct.erl
index 6a3cafbc28..b918d2e671 100644
--- a/src/rabbit_direct.erl
+++ b/src/rabbit_direct.erl
@@ -185,7 +185,8 @@ connect1(User, VHost, Protocol, Pid, Infos) ->
     % Note: peer_host can be either a tuple or
     % a binary if reverse_dns_lookups is enabled
     PeerHost = proplists:get_value(peer_host, Infos),
-    try rabbit_access_control:check_vhost_access(User, VHost, {ip, PeerHost}) of
+    AuthzContext = proplists:get_value(variable_map, Infos, #{}),
+    try rabbit_access_control:check_vhost_access(User, VHost, {ip, PeerHost}, AuthzContext) of
         ok -> ok = pg_local:join(rabbit_direct, Pid),
 	      rabbit_core_metrics:connection_created(Pid, Infos),
               rabbit_event:notify(connection_created, Infos),
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index aa26cf5482..8f64a70b5f 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -1229,7 +1229,7 @@ handle_method0(#'connection.open'{virtual_host = VHost},
                            throttle         = Throttle}) ->
 
     ok = is_over_connection_limit(VHost, User),
-    ok = rabbit_access_control:check_vhost_access(User, VHost, {socket, Sock}),
+    ok = rabbit_access_control:check_vhost_access(User, VHost, {socket, Sock}, #{}),
     ok = is_vhost_alive(VHost, User),
     NewConnection = Connection#connection{vhost = VHost},
     ok = send_on_channel0(Sock, #'connection.open_ok'{}, Protocol),