Merge pull request #16942 from rabbitmq/mk-item-v8qj

michaelklishin · web-flow · commit 0b4eca983ef9 · 2026-07-14T20:59:46.000-07:00
Federation: be more strict when validating URIs
diff --git a/deps/rabbitmq_federation_common/Makefile b/deps/rabbitmq_federation_common/Makefile
@@ -14,7 +14,7 @@ define PROJECT_APP_EXTRA_KEYS
 endef
 
 DEPS = rabbit_common rabbit amqp_client
-TEST_DEPS = rabbitmq_ct_helpers rabbitmq_ct_client_helpers
+TEST_DEPS = rabbitmq_ct_helpers rabbitmq_ct_client_helpers proper
 
 PLT_APPS += rabbitmq_cli
 
diff --git a/deps/rabbitmq_federation_common/src/rabbit_federation_parameters.erl b/deps/rabbitmq_federation_common/src/rabbit_federation_parameters.erl
@@ -10,6 +10,7 @@
 -behaviour(rabbit_policy_validator).
 
 -include("rabbit_federation.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
 
 -export([validate/5, notify/5, notify_clear/4]).
 -export([register/0, unregister/0, validate_policy/1, adjust/1]).
@@ -47,10 +48,10 @@ validate(_VHost, <<"federation-upstream-set">>, Name, Term0, _User) ->
         shared_validation()], Upstream)
      || Upstream <- Term];
 
-validate(_VHost, <<"federation-upstream">>, Name, Term0, _User) ->
+validate(_VHost, <<"federation-upstream">>, Name, Term0, User) ->
     Term = rabbit_data_coercion:to_proplist(Term0),
     rabbit_parameter_validation:proplist(
-      Name, [{<<"uri">>, fun validate_uri/2, mandatory} |
+      Name, [{<<"uri">>, validate_uri(User), mandatory} |
             shared_validation()], Term);
 
 validate(_VHost, _Component, Name, _Term, _User) ->
@@ -95,25 +96,41 @@ shared_validation() ->
      {<<"channel-use-mode">>, rabbit_parameter_validation:enum(
                               ['multiple', 'single']), optional}].
 
-validate_uri(Name, Term) when is_binary(Term) ->
+validate_uri(User) ->
+    fun (Name, Term) -> validate_uri(Name, Term, User) end.
+
+validate_uri(Name, Term, User) when is_binary(Term) ->
     case rabbit_parameter_validation:binary(Name, Term) of
         ok -> case amqp_uri:parse(binary_to_list(Term)) of
-                  {ok, _}    -> ok;
+                  {ok, Params} -> validate_uri_vhost_access(Params, User);
                   {error, E} -> {error, "\"~ts\" not a valid URI: ~tp", [Term, E]}
               end;
         E  -> E
     end;
-validate_uri(Name, Term) ->
+validate_uri(Name, Term, User) ->
     case rabbit_parameter_validation:list(Name, Term) of
         ok -> case [V || U <- Term,
-                         V <- [validate_uri(Name, U)],
+                         V <- [validate_uri(Name, U, User)],
                          element(1, V) =:= error] of
                   []      -> ok;
                   [E | _] -> E
               end;
         E  -> E
     end.
 
+%% A direct URI carries its own virtual host; check access to it here.
+validate_uri_vhost_access(#amqp_params_direct{}, none) ->
+    ok;
+validate_uri_vhost_access(#amqp_params_direct{virtual_host = VHost}, User = #user{}) ->
+    try rabbit_access_control:check_vhost_access(User, VHost, undefined, #{}) of
+        ok -> ok
+    catch
+        _:_ -> {error, "user \"~ts\" may not connect to vhost \"~ts\"",
+                [User#user.username, VHost]}
+    end;
+validate_uri_vhost_access(_Params, _User) ->
+    ok.
+
 %%----------------------------------------------------------------------------
 
 validate_policy([{<<"federation-upstream-set">>, Value}])
diff --git a/deps/rabbitmq_federation_common/test/unit_inbroker_SUITE.erl b/deps/rabbitmq_federation_common/test/unit_inbroker_SUITE.erl
@@ -154,26 +154,26 @@ upstream_set_validation(_Config) ->
                                          <<"a-name">>,
                                           [[{<<"upstream">>, <<"devtest1">>}],
                                            [{<<"upstream">>, <<"devtest2">>}]],
-                                         <<"acting-user">>),
+                                         none),
                  [[ok], [ok]]),
     ?assertEqual(rabbit_federation_parameters:validate(<<"/">>, <<"federation-upstream-set">>,
                                          <<"a-name">>,
                                           [#{<<"upstream">> => <<"devtest3">>},
                                            #{<<"upstream">> => <<"devtest4">>}],
-                                         <<"acting-user">>),
+                                         none),
                  [[ok], [ok]]),
     ok.
 
 upstream_validation(_Config) ->
     ?assertEqual(rabbit_federation_parameters:validate(<<"/">>, <<"federation-upstream">>,
                                          <<"a-name">>,
                                           [{<<"uri">>, <<"amqp://127.0.0.1/%2f">>}],
-                                         <<"acting-user">>),
+                                         none),
                  [ok]),
     ?assertEqual(rabbit_federation_parameters:validate(<<"/">>, <<"federation-upstream">>,
                                          <<"a-name">>,
                                           #{<<"uri">> => <<"amqp://127.0.0.1/%2f">>},
-                                         <<"acting-user">>),
+                                         none),
                  [ok]),
     ok.
 
diff --git a/deps/rabbitmq_federation_common/test/upstream_validation_SUITE.erl b/deps/rabbitmq_federation_common/test/upstream_validation_SUITE.erl
@@ -0,0 +1,180 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-module(upstream_validation_SUITE).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("proper/include/proper.hrl").
+-include_lib("eunit/include/eunit.hrl").
+
+-compile(export_all).
+
+-define(ITERATIONS, 100).
+
+-define(USER, <<"upstream-validation-user">>).
+-define(PASSWORD, <<"upstream-validation-password">>).
+-define(ALLOWED_VHOST, <<"upstream-validation-allowed">>).
+-define(DENIED_VHOST, <<"upstream-validation-denied">>).
+-define(NONEXISTENT_VHOST, <<"upstream-validation-nonexistent">>).
+
+all() ->
+    [
+      {group, tests}
+    ].
+
+groups() ->
+    [
+      {tests, [], [
+          direct_uri_allowed,
+          direct_uri_denied,
+          direct_uri_nonexistent_vhost,
+          direct_uri_no_user,
+          network_uri_allowed,
+          uri_list_denied,
+          uri_list_allowed,
+          map_uri_denied,
+          uri_list_mixed_network_and_denied_direct,
+          prop_uri_list_validation
+        ]}
+    ].
+
+%% -------------------------------------------------------------------
+%% Testsuite setup/teardown
+%% -------------------------------------------------------------------
+
+init_per_suite(Config) ->
+    rabbit_ct_helpers:log_environment(),
+    Config1 = rabbit_ct_helpers:set_config(Config, [{rmq_nodename_suffix, ?MODULE}]),
+    Config2 = rabbit_ct_helpers:run_setup_steps(Config1,
+      rabbit_ct_broker_helpers:setup_steps() ++
+      rabbit_ct_client_helpers:setup_steps()),
+    ok = rabbit_ct_broker_helpers:add_vhost(Config2, ?ALLOWED_VHOST),
+    ok = rabbit_ct_broker_helpers:add_vhost(Config2, ?DENIED_VHOST),
+    ok = rabbit_ct_broker_helpers:add_user(Config2, ?USER, ?PASSWORD),
+    ok = rabbit_ct_broker_helpers:set_full_permissions(Config2, ?USER, ?ALLOWED_VHOST),
+    Config2.
+
+end_per_suite(Config) ->
+    rabbit_ct_broker_helpers:delete_user(Config, ?USER),
+    rabbit_ct_broker_helpers:delete_vhost(Config, ?ALLOWED_VHOST),
+    rabbit_ct_broker_helpers:delete_vhost(Config, ?DENIED_VHOST),
+    rabbit_ct_helpers:run_teardown_steps(Config,
+      rabbit_ct_client_helpers:teardown_steps() ++
+      rabbit_ct_broker_helpers:teardown_steps()).
+
+init_per_group(_, Config) ->
+    Config.
+
+end_per_group(_, Config) ->
+    Config.
+
+init_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_started(Config, Testcase).
+
+end_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_finished(Config, Testcase).
+
+%% -------------------------------------------------------------------
+%% Testcases
+%% -------------------------------------------------------------------
+
+direct_uri_allowed(Config) ->
+    ?assertEqual([ok], validate_with_user(Config, direct_uri(?ALLOWED_VHOST))).
+
+direct_uri_denied(Config) ->
+    ?assertMatch([{error, _, _}], validate_with_user(Config, direct_uri(?DENIED_VHOST))).
+
+direct_uri_nonexistent_vhost(Config) ->
+    ?assertMatch([{error, _, _}], validate_with_user(Config, direct_uri(?NONEXISTENT_VHOST))).
+
+direct_uri_no_user(Config) ->
+    ?assertEqual([ok], validate_with_none(Config, direct_uri(?DENIED_VHOST))).
+
+%% A network URI authenticates over the wire on connect, so it is unchecked here.
+network_uri_allowed(Config) ->
+    ?assertEqual([ok], validate_with_user(Config, network_uri(?DENIED_VHOST))).
+
+uri_list_denied(Config) ->
+    Uris = [direct_uri(?ALLOWED_VHOST), direct_uri(?DENIED_VHOST)],
+    ?assertMatch([{error, _, _}], validate_with_user(Config, Uris)).
+
+uri_list_allowed(Config) ->
+    Uris = [direct_uri(?ALLOWED_VHOST), direct_uri(?ALLOWED_VHOST)],
+    ?assertEqual([ok], validate_with_user(Config, Uris)).
+
+map_uri_denied(Config) ->
+    Term = #{<<"uri">> => direct_uri(?DENIED_VHOST)},
+    ?assertMatch([{error, _, _}], validate_term_with_user(Config, Term)).
+
+uri_list_mixed_network_and_denied_direct(Config) ->
+    Uris = [network_uri(?DENIED_VHOST), direct_uri(?DENIED_VHOST)],
+    ?assertMatch([{error, _, _}], validate_with_user(Config, Uris)).
+
+%% -------------------------------------------------------------------
+%% Property-based test
+%% -------------------------------------------------------------------
+
+%% Runs on the broker node so each example is a local call, not an RPC.
+prop_uri_list_validation(Config) ->
+    rabbit_ct_broker_helpers:rpc(Config, ?MODULE, run_prop_uri_list_validation, []).
+
+run_prop_uri_list_validation() ->
+    {ok, User} = rabbit_access_control:check_user_login(?USER, [{password, ?PASSWORD}]),
+    rabbit_ct_proper_helpers:run_proper(
+      fun() ->
+              ?FORALL(
+                 Choices, non_empty(list(oneof([allowed, denied]))),
+                 begin
+                     Uris = [direct_uri(vhost_for(Choice)) || Choice <- Choices],
+                     Result = validate(Uris, User),
+                     ExpectAllowed = lists:all(fun(C) -> C =:= allowed end, Choices),
+                     outcome(Result) =:= (case ExpectAllowed of
+                                              true  -> allowed;
+                                              false -> denied
+                                          end)
+                 end)
+      end, [], ?ITERATIONS).
+
+outcome([ok]) -> allowed;
+outcome([{error, _, _}]) -> denied.
+
+%% -------------------------------------------------------------------
+%% Helpers
+%% -------------------------------------------------------------------
+
+vhost_for(allowed)     -> ?ALLOWED_VHOST;
+vhost_for(denied)      -> ?DENIED_VHOST;
+vhost_for(nonexistent) -> ?NONEXISTENT_VHOST.
+
+direct_uri(VHost) ->
+    <<"amqp:///", VHost/binary>>.
+
+network_uri(VHost) ->
+    <<"amqp://localhost:5672/", VHost/binary>>.
+
+validate_with_user(Config, Uris) ->
+    validate_term_with_user(Config, [{<<"uri">>, Uris}]).
+
+validate_term_with_user(Config, Term) ->
+    rabbit_ct_broker_helpers:rpc(Config, ?MODULE, validate_term_with_user_1, [Term]).
+
+validate_term_with_user_1(Term) ->
+    {ok, User} = rabbit_access_control:check_user_login(?USER, [{password, ?PASSWORD}]),
+    validate_term(Term, User).
+
+validate_with_none(Config, Uris) ->
+    rabbit_ct_broker_helpers:rpc(Config, ?MODULE, validate_with_none_1, [Uris]).
+
+validate_with_none_1(Uris) ->
+    validate(Uris, none).
+
+validate(Uris, User) ->
+    validate_term([{<<"uri">>, Uris}], User).
+
+validate_term(Term, User) ->
+    rabbit_federation_parameters:validate(
+      <<"/">>, <<"federation-upstream">>, <<"a-name">>, Term, User).
