rabbit_direct.erl

%% 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(rabbit_direct).

-export([boot/0, force_event_refresh/1, list/0, connect/4, connect/5,
         start_channel/9, start_channel/10, disconnect/2]).

-deprecated([{force_event_refresh, 1, eventually}]).

%% Internal
-export([list_local/0,
         conserve_resources/3]).

-include_lib("rabbit_common/include/rabbit.hrl").
-include_lib("rabbit_common/include/rabbit_misc.hrl").
-include_lib("kernel/include/logger.hrl").

%%----------------------------------------------------------------------------

-spec boot() -> 'ok'.

boot() -> rabbit_sup:start_supervisor_child(
            rabbit_direct_client_sup, rabbit_client_sup,
            [{local, rabbit_direct_client_sup},
             {rabbit_channel_sup, start_link, []}]).

-spec force_event_refresh(reference()) -> 'ok'.

force_event_refresh(Ref) ->
    _ = [Pid ! {force_event_refresh, Ref} || Pid <- list()],
    ok.

-spec list_local() -> [pid()].

list_local() ->
    pg_local:get_members(rabbit_direct).

-spec list() -> [pid()].

list() ->
    Nodes = rabbit_nodes:list_running(),
    rabbit_misc:append_rpc_all_nodes(Nodes, rabbit_direct, list_local, [], ?RPC_TIMEOUT).

%%----------------------------------------------------------------------------

auth_fun({none, _}, _VHost, _ExtraAuthProps) ->
    fun () -> {ok, rabbit_auth_backend_dummy:user()} end;

auth_fun({Username, none}, _VHost, ExtraAuthProps) ->
    fun () -> rabbit_access_control:check_user_login(Username, [] ++ ExtraAuthProps) end;

auth_fun({Username, Password}, VHost, ExtraAuthProps) ->
    fun () ->
        rabbit_access_control:check_user_login(
            Username,
            [{password, Password}, {vhost, VHost}] ++ ExtraAuthProps)
    end.

-spec connect
        (({'none', 'none'} | {rabbit_types:username(), 'none'} |
          {rabbit_types:username(), rabbit_types:password()}),
         rabbit_types:vhost(), rabbit_types:protocol(), pid(),
         rabbit_event:event_props()) ->
            rabbit_types:ok_or_error2(
              {rabbit_types:user(), rabbit_framing:amqp_table()},
              'broker_not_found_on_node' |
              {'auth_failure', string()} | 'access_refused').

%% Kept for compatibility with older amqp_client versions (rpc:call).
connect(Creds, VHost, _Protocol, Pid, Infos) ->
    connect(Creds, VHost, Pid, Infos).

-spec connect
        (({'none', 'none'} | {rabbit_types:username(), 'none'} |
          {rabbit_types:username(), rabbit_types:password()}),
         rabbit_types:vhost(), pid(),
         rabbit_event:event_props()) ->
            rabbit_types:ok_or_error2(
              {rabbit_types:user(), rabbit_framing:amqp_table()},
              'broker_not_found_on_node' |
              {'auth_failure', string()} | 'access_refused').

%% Infos is a PropList which contains the content of the Proplist #amqp_adapter_info.additional_info
%% among other credentials such as protocol, ssl information, etc.
%% #amqp_adapter_info.additional_info may carry a credential called `authz_bakends` which has the
%% content of the #user.authz_backends attribute. This means that we are propagating the outcome
%% from the first successful authentication for the current user when opening an internal
%% amqp connection. This is particularly relevant for protocol plugins such as AMQP 1.0 where
%% users are authenticated in one context and later on an internal amqp connection is opened
%% on a different context. In other words, we do not have anymore the initial credentials presented
%% during the first authentication. However, we do have the outcome from such successful authentication.

connect(Creds, VHost, Pid, Infos) ->
    ExtraAuthProps = get_authz_backends(Infos),

    AuthFun = auth_fun(Creds, VHost, ExtraAuthProps),
    case rabbit_boot_state:has_reached_and_is_active(core_started) of
        true  ->
            case whereis(rabbit_direct_client_sup) of
                undefined ->
                    {error, broker_is_booting};
                _ ->
                    case is_over_vhost_connection_limit(VHost, Creds, Pid) of
                        true  ->
                            {error, not_allowed};
                        false ->
                            case is_vhost_alive(VHost, Creds, Pid) of
                                false ->
                                    {error, {internal_error, vhost_is_down}};
                                true  ->
                                    case AuthFun() of
                                        {ok, User = #user{username = Username}} ->
                                            notify_auth_result(Username,
                                                               user_authentication_success, []),
                                            connect1(User, VHost, Pid, Infos);
                                        {refused, Username, Msg, Args} ->
                                            notify_auth_result(Username,
                                                               user_authentication_failure,
                                                               [{error, rabbit_misc:format(Msg, Args)}]),
                                            {error, {auth_failure, "Refused"}}
                                    end %% AuthFun()
                            end %% is_vhost_alive
                    end %% is_over_vhost_connection_limit
            end;
        false -> {error, broker_not_found_on_node}
    end.

get_authz_backends(Infos) ->
    proplists:get_value(authz_backends, Infos, []).

is_vhost_alive(VHost, {Username, _Password}, Pid) ->
    PrintedUsername = case Username of
        none -> "";
        _    -> Username
    end,
    case rabbit_vhost_sup_sup:is_vhost_alive(VHost) of
        true  -> true;
        false ->
            ?LOG_ERROR(
                "Error on direct client connection ~tp~n"
                "access to vhost '~ts' refused for user '~ts': "
                "vhost '~ts' is down",
                [Pid, VHost, PrintedUsername, VHost]),
            false
    end.

is_over_vhost_connection_limit(VHost, {Username, _Password}, Pid) ->
    PrintedUsername = case Username of
        none -> "";
        _    -> Username
    end,
    try rabbit_vhost_limit:is_over_connection_limit(VHost) of
        false         -> false;
        {true, Limit} ->
            ?LOG_ERROR(
                "Error on direct client connection ~tp~n"
                "access to vhost '~ts' refused for user '~ts': "
                "vhost connection limit (~tp) is reached",
                [Pid, VHost, PrintedUsername, Limit]),
            true
    catch
        throw:{error, {no_such_vhost, VHost}} ->
            ?LOG_ERROR(
                "Error on direct client connection ~tp~n"
                "vhost ~ts not found", [Pid, VHost]),
            true
    end.

notify_auth_result(Username, AuthResult, ExtraProps) ->
    EventProps = [{connection_type, direct},
                  {name, case Username of none -> ''; _ -> Username end}] ++
                 ExtraProps,
    rabbit_event:notify(AuthResult, [P || {_, V} = P <- EventProps, V =/= '']).

connect1(User = #user{username = Username}, VHost, Pid, Infos) ->
    case rabbit_auth_backend_internal:is_over_connection_limit(Username) of
        false ->
            % Note: peer_host can be either a tuple or
            % a binary if reverse_dns_lookups is enabled
            PeerHost = proplists:get_value(peer_host, Infos),
            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),
                      _ = rabbit_alarm:register(
                            Pid, {?MODULE, conserve_resources, []}),
                      {ok, {User, rabbit_reader:server_properties(rabbit_framing_amqp_0_9_1)}}
            catch
                exit:#amqp_error{name = Reason = not_allowed} ->
                    {error, Reason}
            end;
        {true, Limit} ->
            ?LOG_ERROR(
                "Error on Direct connection ~tp~n"
                "access refused for user '~ts': "
                "user connection limit (~tp) is reached",
                [Pid, Username, Limit]),
            {error, not_allowed}
    end.

-spec start_channel
        (rabbit_channel:channel_number(), pid(), pid(), string(),
         rabbit_types:protocol(), rabbit_types:user(), rabbit_types:vhost(),
         rabbit_framing:amqp_table(), pid(), any()) ->
            {'ok', pid()}.

%% Kept for compatibility with older amqp_client versions (rpc:call).
start_channel(Number, ClientChannelPid, ConnPid, ConnName, _Protocol,
              User, VHost, Capabilities, Collector, AmqpParams) ->
    start_channel(Number, ClientChannelPid, ConnPid, ConnName,
                  User, VHost, Capabilities, Collector, AmqpParams).

-spec start_channel
        (rabbit_channel:channel_number(), pid(), pid(), string(),
         rabbit_types:user(), rabbit_types:vhost(),
         rabbit_framing:amqp_table(), pid(), any()) ->
            {'ok', pid()}.

start_channel(Number, ClientChannelPid, ConnPid, ConnName,
              User = #user{username = Username}, VHost, Capabilities,
              Collector, AmqpParams) ->
    case rabbit_auth_backend_internal:is_over_channel_limit(Username) of
        false ->
            case rabbit_reader:is_over_node_channel_limit() of
                false ->
                    {ok, _, {ChannelPid, _}} =
                        supervisor:start_child(
                          rabbit_direct_client_sup,
                          [{direct, Number, ClientChannelPid, ConnPid, ConnName,
                            User, VHost, Capabilities, Collector, AmqpParams}]),
                    {ok, ChannelPid};
                {true, NodeLimit} ->
                    ?LOG_ERROR(
                        "Error on direct connection ~tp~n"
                        "number of channels opened on node '~ts' has reached "
                        "the maximum allowed limit of ~w",
                        [ConnPid, node(), NodeLimit]),
                    {error, not_allowed}
            end;
        {true, Limit} ->
            ?LOG_ERROR(
                "Error on direct connection ~tp~n"
                "number of channels opened for user '~ts' has reached the "
                "maximum allowed limit of ~w",
                [ConnPid, Username, Limit]),
            {error, not_allowed}
    end.

-spec disconnect(pid(), rabbit_event:event_props()) -> 'ok'.

disconnect(Pid, Infos) ->
    pg_local:leave(rabbit_direct, Pid),
    rabbit_core_metrics:connection_closed(Pid),
    rabbit_event:notify(connection_closed, Infos).

-spec conserve_resources(pid(),
                         rabbit_alarm:resource_alarm_source(),
                         rabbit_alarm:resource_alert()) -> 'ok'.
conserve_resources(ChannelPid, Source, {_, Conserve, _}) ->
    gen_server:cast(ChannelPid, {conserve_resources, Source, Conserve}).