Khepri: distinguish missing keys from errors in one place

Author: michaelklishin

deps/rabbit/src/rabbit_auth_backend_internal.erl

@@ -163,6 +163,8 @@ check_topic_access(#auth_user{username = Username},
     case rabbit_db_user:get_topic_permissions(Username, VHostPath, Name) of
         undefined ->
             true;
+        {error, _} = Err ->
+            Err;
         #topic_permission{permission = P} ->
             PermRegexp = case element(permission_index(Permission), P) of
                              %% <<"^$">> breaks Emacs' erlang mode

deps/rabbit/src/rabbit_db_user.erl

@@ -404,12 +404,14 @@ clear_all_permissions_for_vhost_in_khepri_tx(VHostName) ->
       Username :: rabbit_types:username(),
       VHostName :: vhost:name(),
       ExchangeName :: binary(),
-      Ret :: TopicPermission | undefined,
+      Ret :: TopicPermission | undefined | {error, term()},
       TopicPermission :: #topic_permission{}.
 %% @doc Returns the topic permissions for the given user and exchange in the
 %% given virtual host.
 %%
-%% @returns the topic permissions record if any, or `undefined'.
+%% @returns the topic permissions record if any, `undefined' if no topic
+%% permissions are set, or `{error, Reason}' if the metadata store query
+%% fails.
 %%
 %% @private
 
@@ -419,8 +421,12 @@ get_topic_permissions(Username, VHostName, ExchangeName)
        is_binary(ExchangeName) ->
     Path = khepri_topic_permission_path(Username, VHostName, ExchangeName),
     case rabbit_khepri:get(Path) of
-        {ok, TopicPermission} -> TopicPermission;
-        _                     -> undefined
+        {ok, TopicPermission} ->
+            TopicPermission;
+        {error, {khepri, node_not_found, _}} ->
+            undefined;
+        {error, _} = Error ->
+            Error
     end.
 
 %% -------------------------------------------------------------------

deps/rabbit/test/topic_permission_SUITE.erl

@@ -7,6 +7,7 @@
 
 -module(topic_permission_SUITE).
 
+-include_lib("proper/include/proper.hrl").
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("amqp10_common/include/amqp10_framing.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
@@ -26,7 +27,9 @@ groups() ->
        amqpl_cc_headers,
        amqpl_bcc_headers,
        topic_permission_database_access,
-       topic_permission_checks
+       topic_permission_checks,
+       topic_permission_khepri_error_fails_closed,
+       topic_permission_khepri_error_fails_closed_prop
       ]}
     ].
 
@@ -393,6 +396,105 @@ topic_permission_checks1(_Config) ->
 
     ok.
 
+%% Topic permission checks must fail closed on metadata store errors.
+topic_permission_khepri_error_fails_closed(Config) ->
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0,
+        ?MODULE, topic_permission_khepri_error_fails_closed1, [Config]).
+
+topic_permission_khepri_error_fails_closed1(_Config) ->
+    clear_tables(),
+    rabbit_vhost:add(<<"/">>, <<"acting-user">>),
+    rabbit_auth_backend_internal:add_user(<<"guest">>, <<"guest">>, <<"acting-user">>),
+    ok = rabbit_auth_backend_internal:set_topic_permissions(
+           <<"guest">>, <<"/">>, <<"amq.topic">>, "^a", "^a", <<"acting-user">>),
+
+    User = #auth_user{username = <<"guest">>},
+    Topic = #resource{name = <<"amq.topic">>, virtual_host = <<"/">>,
+                      kind = topic},
+    Context = #{routing_key => <<"secret.key">>},
+
+    %% Baseline: routing key does not match "^a", so access is denied
+    false = rabbit_auth_backend_internal:check_topic_access(
+              User, Topic, write, Context),
+
+    %% Mock rabbit_khepri:get to simulate a Khepri timeout
+    ok = meck:new(rabbit_khepri, [passthrough]),
+    meck:expect(rabbit_khepri, get,
+                fun(_Path) -> {error, timeout} end),
+    try
+        %% Must return {error, timeout}, not 'true'
+        {error, timeout} = rabbit_auth_backend_internal:check_topic_access(
+                             User, Topic, write, Context),
+        %% Also test with noproc
+        meck:expect(rabbit_khepri, get,
+                    fun(_Path) -> {error, noproc} end),
+        {error, noproc} = rabbit_auth_backend_internal:check_topic_access(
+                            User, Topic, write, Context)
+    after
+        meck:unload(rabbit_khepri)
+    end,
+
+    %% After unmocking, normal behavior resumes
+    false = rabbit_auth_backend_internal:check_topic_access(
+              User, Topic, write, Context),
+    true = rabbit_auth_backend_internal:check_topic_access(
+             User, Topic, write, #{routing_key => <<"a.b.c">>}),
+    ok.
+
+%% Property: for any error reason, check_topic_access must propagate it.
+topic_permission_khepri_error_fails_closed_prop(Config) ->
+    ok = rabbit_ct_broker_helpers:rpc(Config, 0,
+        ?MODULE, topic_permission_khepri_error_fails_closed_prop1, [Config]).
+
+topic_permission_khepri_error_fails_closed_prop1(_Config) ->
+    clear_tables(),
+    rabbit_vhost:add(<<"/">>, <<"acting-user">>),
+    rabbit_auth_backend_internal:add_user(<<"guest">>, <<"guest">>, <<"acting-user">>),
+    ok = rabbit_auth_backend_internal:set_topic_permissions(
+           <<"guest">>, <<"/">>, <<"amq.topic">>, "^a", "^a", <<"acting-user">>),
+    ok = meck:new(rabbit_khepri, [passthrough]),
+    try
+        Property = fun() -> prop_khepri_error_fails_closed() end,
+        rabbit_ct_proper_helpers:run_proper(Property, [], 100)
+    after
+        meck:unload(rabbit_khepri)
+    end,
+    ok.
+
+prop_khepri_error_fails_closed() ->
+    ?FORALL(
+       {ErrorReason, RoutingKey, Permission},
+       {khepri_error_reason(), topic_routing_key(), oneof([read, write])},
+       begin
+           meck:expect(rabbit_khepri, get,
+                       fun(_Path) -> {error, ErrorReason} end),
+           User = #auth_user{username = <<"guest">>},
+           Topic = #resource{name = <<"amq.topic">>, virtual_host = <<"/">>,
+                             kind = topic},
+           Context = #{routing_key => RoutingKey},
+           Result = rabbit_auth_backend_internal:check_topic_access(
+                      User, Topic, Permission, Context),
+           ?WHENFAIL(
+              ct:pal("FAIL: error=~p, routing_key=~p, perm=~p, result=~p",
+                     [ErrorReason, RoutingKey, Permission, Result]),
+              Result =:= {error, ErrorReason})
+       end).
+
+khepri_error_reason() ->
+    oneof([timeout, noproc,
+           {timeout, {gen_server, call, [fake_pid]}},
+           {nodedown, 'node@host'},
+           {no_more_servers_to_try, []},
+           {badmatch, undefined}]).
+
+topic_routing_key() ->
+    oneof([<<"a.b.c">>, <<"secret.key">>, <<"public.1">>,
+           <<"x.y.z">>, <<"">>,
+           ?LET(Parts,
+                non_empty(list(oneof([<<"a">>, <<"b">>, <<"secret">>,
+                                      <<"public">>, <<"x">>]))),
+                iolist_to_binary(lists:join(<<".">>, Parts)))]).
+
 clear_tables() ->
     ok = rabbit_db_vhost:clear(),
     ok = rabbit_db_user:clear().