Support topic authorisation for publishing

References #5
Related to rabbitmq/rabbitmq-server#505 and rabbitmq/rabbitmq-server#1085

deps/rabbitmq_auth_backend_cache/src/rabbit_auth_backend_cache.erl

@@ -21,7 +21,7 @@
 -behaviour(rabbit_authz_backend).
 
 -export([user_login_authentication/2, user_login_authorization/1,
-         check_vhost_access/3, check_resource_access/3]).
+         check_vhost_access/3, check_resource_access/3, check_topic_access/4]).
 
 %% Implementation of rabbit_auth_backend
 
@@ -59,6 +59,15 @@ check_resource_access(#auth_user{} = AuthUser,
            (_)                -> unknown
         end).
 
+check_topic_access(#auth_user{} = AuthUser,
+                   #resource{} = Resource, Permission, Context) ->
+    with_cache(authz, {check_topic_access, [AuthUser, Resource, Permission, Context]},
+        fun(true)  -> success;
+            (false) -> refusal;
+            ({error, _} = Err) -> Err;
+            (_)                -> unknown
+        end).
+
 with_cache(BackendType, {F, A}, Fun) ->
     {ok, AuthCache} = application:get_env(rabbitmq_auth_backend_cache,
                                           cache_module),

deps/rabbitmq_auth_backend_cache/test/src/rabbit_auth_backend_cache_SUITE.erl

@@ -9,7 +9,8 @@ all() ->
     authentication_response,
     authorization_response,
     access_response,
-    cache_expiration
+    cache_expiration,
+    cache_expiration_topic
     ].
 
 init_per_suite(Config) ->
@@ -27,6 +28,27 @@ setup_env(Config) ->
 end_per_suite(Config) ->
     rabbit_ct_helpers:run_teardown_steps(Config, rabbit_ct_broker_helpers:teardown_steps()).
 
+init_per_testcase(access_response, Config) ->
+    ok = rpc(Config, rabbit_auth_backend_internal, set_topic_permissions, [
+        <<"guest">>, <<"/">>, <<"amq.topic">>, <<"^a">>, <<"acting-user">>
+    ]),
+    Config;
+init_per_testcase(_TestCase, Config) ->
+    Config.
+
+end_per_testcase(TestCase, Config) when TestCase == access_response;
+                                        TestCase == cache_expiration_topic ->
+    ok = rpc(Config, rabbit_auth_backend_internal, clear_topic_permissions, [
+        <<"guest">>, <<"/">>, <<"acting-user">>
+    ]),
+    Config;
+end_per_testcase(cache_expiration, Config) ->
+    rabbit_ct_broker_helpers:add_user(Config, <<"guest">>),
+    rabbit_ct_broker_helpers:set_full_permissions(Config, <<"/">>),
+    Config;
+end_per_testcase(_TestCase, Config) ->
+    Config.
+
 authentication_response(Config) ->
     {ok, AuthRespOk} = rpc(Config,rabbit_auth_backend_internal, user_login_authentication, [<<"guest">>, [{password, <<"guest">>}]]),
     {ok, AuthRespOk} = rpc(Config,rabbit_auth_backend_cache, user_login_authentication, [<<"guest">>, [{password, <<"guest">>}]]),
@@ -43,8 +65,11 @@ authorization_response(Config) ->
 access_response(Config) ->
     AvailableVhost = <<"/">>,
     RestrictedVhost = <<"restricted">>,
-    AvailableResource = #resource{virtual_host = AvailableVhost, kind = excahnge, name = <<"some">>},
-    RestrictedResource = #resource{virtual_host = RestrictedVhost, kind = excahnge, name = <<"some">>},
+    AvailableResource = #resource{virtual_host = AvailableVhost, kind = exchange, name = <<"some">>},
+    RestrictedResource = #resource{virtual_host = RestrictedVhost, kind = exchange, name = <<"some">>},
+    TopicResource = #resource{virtual_host = AvailableVhost, kind = topic, name = <<"amq.topic">>},
+    AuthorisedTopicContext = #{routing_key => <<"a.b">>},
+    RestrictedTopicContext = #{routing_key => <<"b.b">>},
 
     {ok, Auth} = rpc(Config,rabbit_auth_backend_internal, user_login_authentication, [<<"guest">>, [{password, <<"guest">>}]]),
     true = rpc(Config,rabbit_auth_backend_internal, check_vhost_access, [Auth, AvailableVhost, none]),
@@ -57,7 +82,13 @@ access_response(Config) ->
     true = rpc(Config,rabbit_auth_backend_cache, check_resource_access, [Auth, AvailableResource, configure]),
 
     false = rpc(Config,rabbit_auth_backend_internal, check_resource_access, [Auth, RestrictedResource, configure]),
-    false = rpc(Config,rabbit_auth_backend_cache, check_resource_access, [Auth, RestrictedResource, configure]).
+    false = rpc(Config,rabbit_auth_backend_cache, check_resource_access, [Auth, RestrictedResource, configure]),
+
+    true = rpc(Config,rabbit_auth_backend_internal, check_topic_access, [Auth, TopicResource, write, AuthorisedTopicContext]),
+    true = rpc(Config,rabbit_auth_backend_cache, check_topic_access, [Auth, TopicResource, write, AuthorisedTopicContext]),
+
+    false = rpc(Config,rabbit_auth_backend_internal, check_topic_access, [Auth, TopicResource, write, RestrictedTopicContext]),
+    false = rpc(Config,rabbit_auth_backend_cache, check_topic_access, [Auth, TopicResource, write, RestrictedTopicContext]).
 
 cache_expiration(Config) ->
     AvailableVhost = <<"/">>,
@@ -99,6 +130,29 @@ cache_expiration(Config) ->
     false = rpc(Config,rabbit_auth_backend_internal, check_resource_access, [Auth, AvailableResource, configure]),
     false = rpc(Config,rabbit_auth_backend_cache, check_resource_access, [Auth, AvailableResource, configure]).
 
+cache_expiration_topic(Config) ->
+    AvailableVhost = <<"/">>,
+    TopicResource = #resource{virtual_host = AvailableVhost, kind = topic, name = <<"amq.topic">>},
+    RestrictedTopicContext = #{routing_key => <<"b.b">>},
+
+    {ok, Auth} = rpc(Config,rabbit_auth_backend_internal, user_login_authentication, [<<"guest">>, [{password, <<"guest">>}]]),
+
+    % topic access is authorised if no permission is found
+    true = rpc(Config,rabbit_auth_backend_internal, check_topic_access, [Auth, TopicResource, write, RestrictedTopicContext]),
+    true = rpc(Config,rabbit_auth_backend_cache, check_topic_access, [Auth, TopicResource, write, RestrictedTopicContext]),
+
+    ok = rpc(Config, rabbit_auth_backend_internal, set_topic_permissions, [
+        <<"guest">>, <<"/">>, <<"amq.topic">>, <<"^a">>, <<"acting-user">>
+    ]),
+
+    false = rpc(Config,rabbit_auth_backend_internal, check_topic_access, [Auth, TopicResource, write, RestrictedTopicContext]),
+    true = rpc(Config,rabbit_auth_backend_cache, check_topic_access, [Auth, TopicResource, write, RestrictedTopicContext]),
+
+    {ok, TTL} = rpc(Config, application, get_env, [rabbitmq_auth_backend_cache, cache_ttl]),
+    timer:sleep(TTL),
+
+    false = rpc(Config,rabbit_auth_backend_internal, check_topic_access, [Auth, TopicResource, write, RestrictedTopicContext]),
+    false = rpc(Config,rabbit_auth_backend_cache, check_topic_access, [Auth, TopicResource, write, RestrictedTopicContext]).
 
 rpc(Config, M, F, A) ->
     rabbit_ct_broker_helpers:rpc(Config, 0, M, F, A).