processone / ejabberd / 538

pending completion

Build # 538

Build Type

push

github

Committed by

weiss

Commit Message

make-binaries: Don't build Linux-PAM examples

Building the examples shipped with Linux-PAM 1.5.3 fails with musl libc:

https://github.com/linux-pam/linux-pam/commit/5374f677e

Run Details

13553 of 41048 relevant lines covered (33.02%)

677.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

48.68

/src/ejabberd_s2s_out.erl

%%%-------------------------------------------------------------------
%%% Created : 16 Dec 2016 by Evgeny Khramtsov <ekhramtsov@process-one.net>
%%%
%%%
%%% ejabberd, Copyright (C) 2002-2023   ProcessOne
%%%
%%% This program is free software; you can redistribute it and/or
%%% modify it under the terms of the GNU General Public License as
%%% published by the Free Software Foundation; either version 2 of the
%%% License, or (at your option) any later version.
%%%
%%% This program is distributed in the hope that it will be useful,
%%% but WITHOUT ANY WARRANTY; without even the implied warranty of
%%% MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
%%% General Public License for more details.
%%%
%%% You should have received a copy of the GNU General Public License along
%%% with this program; if not, write to the Free Software Foundation, Inc.,
%%% 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
%%%
%%%-------------------------------------------------------------------
-module(ejabberd_s2s_out).
-behaviour(xmpp_stream_out).

%% xmpp_stream_out callbacks
-export([tls_options/1, tls_required/1, tls_verify/1, tls_enabled/1,
         connect_options/3, connect_timeout/1, address_families/1, default_port/1,
         dns_retries/1, dns_timeout/1,
         handle_auth_success/2, handle_auth_failure/3, handle_packet/2,
         handle_stream_end/2, handle_stream_downgraded/2,
         handle_recv/3, handle_send/3, handle_cdata/2,
         handle_stream_established/1, handle_timeout/1]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
         terminate/2, code_change/3]).
%% Hooks
-export([process_auth_result/2, process_closed/2, handle_unexpected_info/2,
         handle_unexpected_cast/2, process_downgraded/2]).
%% API
-export([start/3, start_link/3, connect/1, close/1, close/2, stop_async/1, send/2,
         route/2, establish/1, update_state/2, host_up/1, host_down/1]).

-include_lib("xmpp/include/xmpp.hrl").
-include("logger.hrl").
-include("translate.hrl").

-type state() :: xmpp_stream_out:state().
-export_type([state/0]).

%%%===================================================================
%%% API
%%%===================================================================
start(From, To, Opts) ->
    case proplists:get_value(supervisor, Opts, true) of
        true ->
            case supervisor:start_child(ejabberd_s2s_out_sup,
                                        [From, To, Opts]) of
                {ok, undefined} -> ignore;
                Res -> Res
            end;
        _ ->
            xmpp_stream_out:start(?MODULE, [From, To, Opts],
                                  ejabberd_config:fsm_limit_opts([]))
    end.

start_link(From, To, Opts) ->
    xmpp_stream_out:start_link(?MODULE, [From, To, Opts],
                               ejabberd_config:fsm_limit_opts([])).

-spec connect(pid()) -> ok.
connect(Ref) ->
    xmpp_stream_out:connect(Ref).

-spec close(pid()) -> ok;
           (state()) -> state().
close(Ref) ->
    xmpp_stream_out:close(Ref).

-spec close(pid(), atom()) -> ok.
close(Ref, Reason) ->
    xmpp_stream_out:close(Ref, Reason).

-spec stop_async(pid()) -> ok.
stop_async(Pid) ->
    xmpp_stream_out:stop_async(Pid).

-spec send(pid(), xmpp_element()) -> ok;
          (state(), xmpp_element()) -> state().
send(Stream, Pkt) ->
    xmpp_stream_out:send(Stream, Pkt).

-spec route(pid(), xmpp_element()) -> ok.
route(Ref, Pkt) ->
    Ref ! {route, Pkt},
    ok.

-spec establish(state()) -> state().
establish(State) ->
    xmpp_stream_out:establish(State).

-spec update_state(pid(), fun((state()) -> state()) |
                   {module(), atom(), list()}) -> ok.
update_state(Ref, Callback) ->
    xmpp_stream_out:cast(Ref, {update_state, Callback}).

-spec host_up(binary()) -> ok.
host_up(Host) ->
    ejabberd_hooks:add(s2s_out_auth_result, Host, ?MODULE,
                       process_auth_result, 100),
    ejabberd_hooks:add(s2s_out_closed, Host, ?MODULE,
                       process_closed, 100),
    ejabberd_hooks:add(s2s_out_handle_info, Host, ?MODULE,
                       handle_unexpected_info, 100),
    ejabberd_hooks:add(s2s_out_handle_cast, Host, ?MODULE,
                       handle_unexpected_cast, 100),
    ejabberd_hooks:add(s2s_out_downgraded, Host, ?MODULE,
                       process_downgraded, 100).

-spec host_down(binary()) -> ok.
host_down(Host) ->
    ejabberd_hooks:delete(s2s_out_auth_result, Host, ?MODULE,
                          process_auth_result, 100),
    ejabberd_hooks:delete(s2s_out_closed, Host, ?MODULE,
                          process_closed, 100),
    ejabberd_hooks:delete(s2s_out_handle_info, Host, ?MODULE,
                          handle_unexpected_info, 100),
    ejabberd_hooks:delete(s2s_out_handle_cast, Host, ?MODULE,
                          handle_unexpected_cast, 100),
    ejabberd_hooks:delete(s2s_out_downgraded, Host, ?MODULE,
                          process_downgraded, 100).

%%%===================================================================
%%% Hooks
%%%===================================================================
process_auth_result(#{server := LServer, remote_server := RServer} = State,
                    {false, Reason}) ->
    Delay = get_delay(),
    ?WARNING_MSG("Failed to establish outbound s2s connection ~ts -> ~ts: "
                 "authentication failed; bouncing for ~p seconds",
                 [LServer, RServer, Delay div 1000]),
    State1 = State#{on_route => bounce, stop_reason => Reason},
    State2 = close(State1),
    State3 = bounce_queue(State2),
    xmpp_stream_out:set_timeout(State3, Delay);
process_auth_result(State, true) ->
    State.

process_closed(#{server := LServer, remote_server := RServer,
                 on_route := send} = State,
               Reason) ->
    ?INFO_MSG("Closing outbound s2s connection ~ts -> ~ts: ~ts",
              [LServer, RServer, format_error(Reason)]),
    stop_async(self()),
    State;
process_closed(#{server := LServer, remote_server := RServer} = State,
               Reason) ->
    Delay = get_delay(),
    ?WARNING_MSG("Failed to establish outbound s2s connection ~ts -> ~ts: ~ts; "
                 "bouncing for ~p seconds",
                 [LServer, RServer, format_error(Reason), Delay div 1000]),
    State1 = State#{on_route => bounce},
    State2 = bounce_queue(State1),
    xmpp_stream_out:set_timeout(State2, Delay).

handle_unexpected_info(State, Info) ->
    ?WARNING_MSG("Unexpected info: ~p", [Info]),
    State.

handle_unexpected_cast(State, Msg) ->
    ?WARNING_MSG("Unexpected cast: ~p", [Msg]),
    State.

process_downgraded(State, _StreamStart) ->
    send(State, xmpp:serr_unsupported_version()).

%%%===================================================================
%%% xmpp_stream_out callbacks
%%%===================================================================
tls_options(#{server := LServer, server_host := ServerHost}) ->
    ejabberd_s2s:tls_options(LServer, ServerHost, []).

tls_required(#{server_host := ServerHost}) ->
    ejabberd_s2s:tls_required(ServerHost).

tls_verify(#{server_host := ServerHost} = State) ->
    ejabberd_hooks:run_fold(s2s_out_tls_verify, ServerHost, true, [State]).

tls_enabled(#{server_host := ServerHost}) ->
    ejabberd_s2s:tls_enabled(ServerHost).

connect_options(Addr, Opts, #{server_host := ServerHost}) ->
    BindAddr = case get_addr_type(Addr) of
                   inet ->
                       ejabberd_option:outgoing_s2s_ipv4_address(ServerHost);
                   inet6 ->
                       ejabberd_option:outgoing_s2s_ipv6_address(ServerHost)
               end,
    case BindAddr of
        undefined ->
            Opts;
        _ ->
            [{ip, BindAddr} | Opts]
    end.

connect_timeout(#{server_host := ServerHost}) ->
    ejabberd_option:outgoing_s2s_timeout(ServerHost).

default_port(#{server_host := ServerHost}) ->
    ejabberd_option:outgoing_s2s_port(ServerHost).

address_families(#{server_host := ServerHost}) ->
    ejabberd_option:outgoing_s2s_families(ServerHost).

dns_retries(#{server_host := ServerHost}) ->
    ejabberd_option:s2s_dns_retries(ServerHost).

dns_timeout(#{server_host := ServerHost}) ->
    ejabberd_option:s2s_dns_timeout(ServerHost).

handle_auth_success(Mech, #{socket := Socket, ip := IP,
                            remote_server := RServer,
                            server_host := ServerHost,
                            server := LServer} = State) ->
    ?INFO_MSG("(~ts) Accepted outbound s2s ~ts authentication ~ts -> ~ts (~ts)",
              [xmpp_socket:pp(Socket), Mech, LServer, RServer,
               ejabberd_config:may_hide_data(misc:ip_to_list(IP))]),
    ejabberd_hooks:run_fold(s2s_out_auth_result, ServerHost, State, [true]).

handle_auth_failure(Mech, Reason,
                    #{socket := Socket, ip := IP,
                      remote_server := RServer,
                      server_host := ServerHost,
                      server := LServer} = State) ->
    ?WARNING_MSG("(~ts) Failed outbound s2s ~ts authentication ~ts -> ~ts (~ts): ~ts",
                 [xmpp_socket:pp(Socket), Mech, LServer, RServer,
                  ejabberd_config:may_hide_data(misc:ip_to_list(IP)),
                  xmpp_stream_out:format_error(Reason)]),
    ejabberd_hooks:run_fold(s2s_out_auth_result, ServerHost, State, [{false, Reason}]).

handle_packet(Pkt, #{server_host := ServerHost} = State) ->
    ejabberd_hooks:run_fold(s2s_out_packet, ServerHost, State, [Pkt]).

handle_stream_end(Reason, #{server_host := ServerHost} = State) ->
    State1 = State#{stop_reason => Reason},
    ejabberd_hooks:run_fold(s2s_out_closed, ServerHost, State1, [Reason]).

handle_stream_downgraded(StreamStart, #{server_host := ServerHost} = State) ->
    ejabberd_hooks:run_fold(s2s_out_downgraded, ServerHost, State, [StreamStart]).

handle_stream_established(State) ->
    State1 = State#{on_route => send},
    State2 = resend_queue(State1),
    set_idle_timeout(State2).

handle_cdata(Data, #{server_host := ServerHost} = State) ->
    ejabberd_hooks:run_fold(s2s_out_handle_cdata, ServerHost, State, [Data]).

handle_recv(El, Pkt, #{server_host := ServerHost} = State) ->
    ejabberd_hooks:run_fold(s2s_out_handle_recv, ServerHost, State, [El, Pkt]).

handle_send(El, Pkt, #{server_host := ServerHost} = State) ->
    ejabberd_hooks:run_fold(s2s_out_handle_send, ServerHost, State, [El, Pkt]).

handle_timeout(#{on_route := Action, lang := Lang} = State) ->
    case Action of
        bounce ->
            stop_async(self()),
            State;
        _ ->
            Txt = ?T("Idle connection"),
            send(State, xmpp:serr_connection_timeout(Txt, Lang))
    end.

init([#{server := LServer, remote_server := RServer} = State, Opts]) ->
    ServerHost = ejabberd_router:host_of_route(LServer),
    QueueType = ejabberd_s2s:queue_type(ServerHost),
    QueueLimit = case lists:keyfind(
                        max_queue, 1, ejabberd_config:fsm_limit_opts([])) of
                     {_, N} -> N;
                     false -> unlimited
                 end,
    Timeout = ejabberd_option:negotiation_timeout(),
    State1 = State#{on_route => queue,
                    queue => p1_queue:new(QueueType, QueueLimit),
                    xmlns => ?NS_SERVER,
                    lang => ejabberd_option:language(),
                    server_host => ServerHost,
                    shaper => none},
    State2 = xmpp_stream_out:set_timeout(State1, Timeout),
    ?INFO_MSG("Outbound s2s connection started: ~ts -> ~ts",
              [LServer, RServer]),
    ejabberd_hooks:run_fold(s2s_out_init, ServerHost, {ok, State2}, [Opts]).

handle_call(Request, From, #{server_host := ServerHost} = State) ->
    ejabberd_hooks:run_fold(s2s_out_handle_call, ServerHost, State, [Request, From]).

handle_cast({update_state, Fun}, State) ->
    case Fun of
        {M, F, A} -> erlang:apply(M, F, [State|A]);
        _ when is_function(Fun) -> Fun(State)
          end;
handle_cast(Msg, #{server_host := ServerHost} = State) ->
    ejabberd_hooks:run_fold(s2s_out_handle_cast, ServerHost, State, [Msg]).

handle_info({route, Pkt}, #{queue := Q, on_route := Action} = State) ->
    case Action of
        queue ->
            try State#{queue => p1_queue:in(Pkt, Q)}
            catch error:full ->
                    Q1 = p1_queue:set_limit(Q, unlimited),
                    Q2 = p1_queue:in(Pkt, Q1),
                    handle_stream_end(queue_full, State#{queue => Q2})
            end;
        bounce -> bounce_packet(Pkt, State);
        send -> set_idle_timeout(send(State, Pkt))
    end;
handle_info(Info, #{server_host := ServerHost} = State) ->
    ejabberd_hooks:run_fold(s2s_out_handle_info, ServerHost, State, [Info]).

terminate(Reason, #{server := LServer,
                    remote_server := RServer} = State) ->
    State1 = case Reason of
                 normal -> State;
                 _ -> State#{stop_reason => internal_failure}
    end,
    State2 = bounce_queue(State1),
    bounce_message_queue({LServer, RServer}, State2).

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%%===================================================================
%%% Internal functions
%%%===================================================================
-spec get_addr_type(inet:ip_address()) -> inet:address_family().
get_addr_type({_, _, _, _}) -> inet;
get_addr_type({_, _, _, _, _, _, _, _}) -> inet6.

-spec resend_queue(state()) -> state().
resend_queue(State) ->
    queue_fold(
      fun(Pkt, AccState) ->
              send(AccState, Pkt)
      end, State).

-spec bounce_queue(state()) -> state().
bounce_queue(State) ->
    queue_fold(
      fun(Pkt, AccState) ->
              bounce_packet(Pkt, AccState)
      end, State).

-spec bounce_message_queue({binary(), binary()}, state()) -> state().
bounce_message_queue(FromTo, State) ->
    receive {route, Pkt} ->
            State1 = bounce_packet(Pkt, State),
            bounce_message_queue(FromTo, State1)
    after 0 ->
            State
    end.

-spec bounce_packet(xmpp_element(), state()) -> state().
bounce_packet(Pkt, State) when ?is_stanza(Pkt) ->
    #{server_host := Host} = State,
    case ejabberd_hooks:run_fold(
           s2s_out_bounce_packet, Host, State, [Pkt]) of
        ignore ->
            State;
        State2 ->
            Lang = xmpp:get_lang(Pkt),
            Err = mk_bounce_error(Lang, State2),
            ejabberd_router:route_error(Pkt, Err),
            State2
    end;
bounce_packet(_, State) ->
    State.

-spec mk_bounce_error(binary(), state()) -> stanza_error().
mk_bounce_error(Lang, #{stop_reason := Why}) ->
    Reason = format_error(Why),
    case Why of
        internal_failure ->
            xmpp:err_internal_server_error(Reason, Lang);
        queue_full ->
            xmpp:err_resource_constraint(Reason, Lang);
        {dns, _} ->
            xmpp:err_remote_server_not_found(Reason, Lang);
        {idna, _} ->
            xmpp:err_remote_server_not_found(Reason, Lang);
        _ ->
            xmpp:err_remote_server_timeout(Reason, Lang)
    end;
mk_bounce_error(_Lang, _State) ->
    %% We should not be here. Probably :)
    xmpp:err_remote_server_not_found().

-spec get_delay() -> non_neg_integer().
get_delay() ->
    MaxDelay = ejabberd_option:s2s_max_retry_delay(),
    p1_rand:uniform(MaxDelay).

-spec set_idle_timeout(state()) -> state().
set_idle_timeout(#{on_route := send, server_host := ServerHost} = State) ->
    Timeout = ejabberd_s2s:get_idle_timeout(ServerHost),
    xmpp_stream_out:set_timeout(State, Timeout);
set_idle_timeout(State) ->
    State.

-spec queue_fold(fun((xmpp_element(), state()) -> state()), state()) -> state().
queue_fold(F, #{queue := Q} = State) ->
    case p1_queue:out(Q) of
        {{value, Pkt}, Q1} ->
            State1 = F(Pkt, State#{queue => Q1}),
            queue_fold(F, State1);
        {empty, Q1} ->
            State#{queue => Q1}
    end.

format_error(internal_failure) ->
    <<"Internal server error">>;
format_error(queue_full) ->
    <<"Stream queue is overloaded">>;
format_error(Reason) ->
    xmpp_stream_out:format_error(Reason).

1	%%%-------------------------------------------------------------------
2	%%% Created : 16 Dec 2016 by Evgeny Khramtsov <ekhramtsov@process-one.net>
3	%%%
4	%%%
5	%%% ejabberd, Copyright (C) 2002-2023 ProcessOne
6	%%%
7	%%% This program is free software; you can redistribute it and/or
8	%%% modify it under the terms of the GNU General Public License as
9	%%% published by the Free Software Foundation; either version 2 of the
10	%%% License, or (at your option) any later version.
11	%%%
12	%%% This program is distributed in the hope that it will be useful,
13	%%% but WITHOUT ANY WARRANTY; without even the implied warranty of
14	%%% MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15	%%% General Public License for more details.
16	%%%
17	%%% You should have received a copy of the GNU General Public License along
18	%%% with this program; if not, write to the Free Software Foundation, Inc.,
19	%%% 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
20	%%%
21	%%%-------------------------------------------------------------------
22	-module(ejabberd_s2s_out).
23	-behaviour(xmpp_stream_out).
24
25	%% xmpp_stream_out callbacks
26	-export([tls_options/1, tls_required/1, tls_verify/1, tls_enabled/1,
27	connect_options/3, connect_timeout/1, address_families/1, default_port/1,
28	dns_retries/1, dns_timeout/1,
29	handle_auth_success/2, handle_auth_failure/3, handle_packet/2,
30	handle_stream_end/2, handle_stream_downgraded/2,
31	handle_recv/3, handle_send/3, handle_cdata/2,
32	handle_stream_established/1, handle_timeout/1]).
33	-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
34	terminate/2, code_change/3]).
35	%% Hooks
36	-export([process_auth_result/2, process_closed/2, handle_unexpected_info/2,
37	handle_unexpected_cast/2, process_downgraded/2]).
38	%% API
39	-export([start/3, start_link/3, connect/1, close/1, close/2, stop_async/1, send/2,
40	route/2, establish/1, update_state/2, host_up/1, host_down/1]).
41
42	-include_lib("xmpp/include/xmpp.hrl").
43	-include("logger.hrl").
44	-include("translate.hrl").
45
46	-type state() :: xmpp_stream_out:state().
47	-export_type([state/0]).
48
49	%%%===================================================================
50	%%% API
51	%%%===================================================================
52	start(From, To, Opts) ->
53	case proplists:get_value(supervisor, Opts, true) of	5✔
54	true ->
55	case supervisor:start_child(ejabberd_s2s_out_sup,	5✔
56	[From, To, Opts]) of
57	{ok, undefined} -> ignore;	×
58	Res -> Res	5✔
59	end;
60	_ ->
61	xmpp_stream_out:start(?MODULE, [From, To, Opts],	×
62	ejabberd_config:fsm_limit_opts([]))
63	end.
64
65	start_link(From, To, Opts) ->
66	xmpp_stream_out:start_link(?MODULE, [From, To, Opts],	5✔
67	ejabberd_config:fsm_limit_opts([])).
68
69	-spec connect(pid()) -> ok.
70	connect(Ref) ->
71	xmpp_stream_out:connect(Ref).	5✔
72
73	-spec close(pid()) -> ok;
74	(state()) -> state().
75	close(Ref) ->
76	xmpp_stream_out:close(Ref).	×
77
78	-spec close(pid(), atom()) -> ok.
79	close(Ref, Reason) ->
80	xmpp_stream_out:close(Ref, Reason).	×
81
82	-spec stop_async(pid()) -> ok.
83	stop_async(Pid) ->
84	xmpp_stream_out:stop_async(Pid).	4✔
85
86	-spec send(pid(), xmpp_element()) -> ok;
87	(state(), xmpp_element()) -> state().
88	send(Stream, Pkt) ->
89	xmpp_stream_out:send(Stream, Pkt).	9✔
90
91	-spec route(pid(), xmpp_element()) -> ok.
92	route(Ref, Pkt) ->
93	Ref ! {route, Pkt},	3✔
94	ok.	3✔
95
96	-spec establish(state()) -> state().
97	establish(State) ->
98	xmpp_stream_out:establish(State).	2✔
99
100	-spec update_state(pid(), fun((state()) -> state()) \|
101	{module(), atom(), list()}) -> ok.
102	update_state(Ref, Callback) ->
103	xmpp_stream_out:cast(Ref, {update_state, Callback}).	×
104
105	-spec host_up(binary()) -> ok.
106	host_up(Host) ->
107	ejabberd_hooks:add(s2s_out_auth_result, Host, ?MODULE,	9✔
108	process_auth_result, 100),
109	ejabberd_hooks:add(s2s_out_closed, Host, ?MODULE,	9✔
110	process_closed, 100),
111	ejabberd_hooks:add(s2s_out_handle_info, Host, ?MODULE,	9✔
112	handle_unexpected_info, 100),
113	ejabberd_hooks:add(s2s_out_handle_cast, Host, ?MODULE,	9✔
114	handle_unexpected_cast, 100),
115	ejabberd_hooks:add(s2s_out_downgraded, Host, ?MODULE,	9✔
116	process_downgraded, 100).
117
118	-spec host_down(binary()) -> ok.
119	host_down(Host) ->
120	ejabberd_hooks:delete(s2s_out_auth_result, Host, ?MODULE,	×
121	process_auth_result, 100),
122	ejabberd_hooks:delete(s2s_out_closed, Host, ?MODULE,	×
123	process_closed, 100),
124	ejabberd_hooks:delete(s2s_out_handle_info, Host, ?MODULE,	×
125	handle_unexpected_info, 100),
126	ejabberd_hooks:delete(s2s_out_handle_cast, Host, ?MODULE,	×
127	handle_unexpected_cast, 100),
128	ejabberd_hooks:delete(s2s_out_downgraded, Host, ?MODULE,	×
129	process_downgraded, 100).
130
131	%%%===================================================================
132	%%% Hooks
133	%%%===================================================================
134	process_auth_result(#{server := LServer, remote_server := RServer} = State,
135	{false, Reason}) ->
136	Delay = get_delay(),	×
137	?WARNING_MSG("Failed to establish outbound s2s connection ~ts -> ~ts: "	×
138	"authentication failed; bouncing for ~p seconds",
139	[LServer, RServer, Delay div 1000]),	×
140	State1 = State#{on_route => bounce, stop_reason => Reason},	×
141	State2 = close(State1),	×
142	State3 = bounce_queue(State2),	×
143	xmpp_stream_out:set_timeout(State3, Delay);	×
144	process_auth_result(State, true) ->
145	State.	3✔
146
147	process_closed(#{server := LServer, remote_server := RServer,
148	on_route := send} = State,
149	Reason) ->
150	?INFO_MSG("Closing outbound s2s connection ~ts -> ~ts: ~ts",	1✔
151	[LServer, RServer, format_error(Reason)]),	×
152	stop_async(self()),	×
153	State;	×
154	process_closed(#{server := LServer, remote_server := RServer} = State,
155	Reason) ->
156	Delay = get_delay(),	×
157	?WARNING_MSG("Failed to establish outbound s2s connection ~ts -> ~ts: ~ts; "	×
158	"bouncing for ~p seconds",
159	[LServer, RServer, format_error(Reason), Delay div 1000]),	×
160	State1 = State#{on_route => bounce},	×
161	State2 = bounce_queue(State1),	×
162	xmpp_stream_out:set_timeout(State2, Delay).	×
163
164	handle_unexpected_info(State, Info) ->
165	?WARNING_MSG("Unexpected info: ~p", [Info]),	×
166	State.	×
167
168	handle_unexpected_cast(State, Msg) ->
169	?WARNING_MSG("Unexpected cast: ~p", [Msg]),	×
170	State.	×
171
172	process_downgraded(State, _StreamStart) ->
173	send(State, xmpp:serr_unsupported_version()).	×
174
175	%%%===================================================================
176	%%% xmpp_stream_out callbacks
177	%%%===================================================================
178	tls_options(#{server := LServer, server_host := ServerHost}) ->
179	ejabberd_s2s:tls_options(LServer, ServerHost, []).	3✔
180
181	tls_required(#{server_host := ServerHost}) ->
182	ejabberd_s2s:tls_required(ServerHost).	8✔
183
184	tls_verify(#{server_host := ServerHost} = State) ->
185	ejabberd_hooks:run_fold(s2s_out_tls_verify, ServerHost, true, [State]).	3✔
186
187	tls_enabled(#{server_host := ServerHost}) ->
188	ejabberd_s2s:tls_enabled(ServerHost).	13✔
189
190	connect_options(Addr, Opts, #{server_host := ServerHost}) ->
191	BindAddr = case get_addr_type(Addr) of	10✔
192	inet ->
193	ejabberd_option:outgoing_s2s_ipv4_address(ServerHost);	5✔
194	inet6 ->
195	ejabberd_option:outgoing_s2s_ipv6_address(ServerHost)	5✔
196	end,
197	case BindAddr of	10✔
198	undefined ->
199	Opts;	10✔
200	_ ->
201	[{ip, BindAddr} \| Opts]	×
202	end.
203
204	connect_timeout(#{server_host := ServerHost}) ->
205	ejabberd_option:outgoing_s2s_timeout(ServerHost).	10✔
206
207	default_port(#{server_host := ServerHost}) ->
208	ejabberd_option:outgoing_s2s_port(ServerHost).	5✔
209
210	address_families(#{server_host := ServerHost}) ->
211	ejabberd_option:outgoing_s2s_families(ServerHost).	×
212
213	dns_retries(#{server_host := ServerHost}) ->
214	ejabberd_option:s2s_dns_retries(ServerHost).	8✔
215
216	dns_timeout(#{server_host := ServerHost}) ->
217	ejabberd_option:s2s_dns_timeout(ServerHost).	10✔
218
219	handle_auth_success(Mech, #{socket := Socket, ip := IP,
220	remote_server := RServer,
221	server_host := ServerHost,
222	server := LServer} = State) ->
223	?INFO_MSG("(~ts) Accepted outbound s2s ~ts authentication ~ts -> ~ts (~ts)",	3✔
224	[xmpp_socket:pp(Socket), Mech, LServer, RServer,
225	ejabberd_config:may_hide_data(misc:ip_to_list(IP))]),	3✔
226	ejabberd_hooks:run_fold(s2s_out_auth_result, ServerHost, State, [true]).	3✔
227
228	handle_auth_failure(Mech, Reason,
229	#{socket := Socket, ip := IP,
230	remote_server := RServer,
231	server_host := ServerHost,
232	server := LServer} = State) ->
233	?WARNING_MSG("(~ts) Failed outbound s2s ~ts authentication ~ts -> ~ts (~ts): ~ts",	4✔
234	[xmpp_socket:pp(Socket), Mech, LServer, RServer,
235	ejabberd_config:may_hide_data(misc:ip_to_list(IP)),
236	xmpp_stream_out:format_error(Reason)]),	4✔
237	ejabberd_hooks:run_fold(s2s_out_auth_result, ServerHost, State, [{false, Reason}]).	4✔
238
239	handle_packet(Pkt, #{server_host := ServerHost} = State) ->
240	ejabberd_hooks:run_fold(s2s_out_packet, ServerHost, State, [Pkt]).	4✔
241
242	handle_stream_end(Reason, #{server_host := ServerHost} = State) ->
243	State1 = State#{stop_reason => Reason},	1✔
244	ejabberd_hooks:run_fold(s2s_out_closed, ServerHost, State1, [Reason]).	1✔
245
246	handle_stream_downgraded(StreamStart, #{server_host := ServerHost} = State) ->
247	ejabberd_hooks:run_fold(s2s_out_downgraded, ServerHost, State, [StreamStart]).	×
248
249	handle_stream_established(State) ->
250	State1 = State#{on_route => send},	3✔
251	State2 = resend_queue(State1),	3✔
252	set_idle_timeout(State2).	3✔
253
254	handle_cdata(Data, #{server_host := ServerHost} = State) ->
255	ejabberd_hooks:run_fold(s2s_out_handle_cdata, ServerHost, State, [Data]).	×
256
257	handle_recv(El, Pkt, #{server_host := ServerHost} = State) ->
258	ejabberd_hooks:run_fold(s2s_out_handle_recv, ServerHost, State, [El, Pkt]).	19✔
259
260	handle_send(El, Pkt, #{server_host := ServerHost} = State) ->
261	ejabberd_hooks:run_fold(s2s_out_handle_send, ServerHost, State, [El, Pkt]).	13✔
262
263	handle_timeout(#{on_route := Action, lang := Lang} = State) ->
264	case Action of	×
265	bounce ->
266	stop_async(self()),	×
267	State;	×
268	_ ->
269	Txt = ?T("Idle connection"),	×
270	send(State, xmpp:serr_connection_timeout(Txt, Lang))	×
271	end.
272
273	init([#{server := LServer, remote_server := RServer} = State, Opts]) ->
274	ServerHost = ejabberd_router:host_of_route(LServer),	5✔
275	QueueType = ejabberd_s2s:queue_type(ServerHost),	5✔
276	QueueLimit = case lists:keyfind(	5✔
277	max_queue, 1, ejabberd_config:fsm_limit_opts([])) of
278	{_, N} -> N;	5✔
279	false -> unlimited	×
280	end,
281	Timeout = ejabberd_option:negotiation_timeout(),	5✔
282	State1 = State#{on_route => queue,	5✔
283	queue => p1_queue:new(QueueType, QueueLimit),
284	xmlns => ?NS_SERVER,
285	lang => ejabberd_option:language(),
286	server_host => ServerHost,
287	shaper => none},
288	State2 = xmpp_stream_out:set_timeout(State1, Timeout),	5✔
289	?INFO_MSG("Outbound s2s connection started: ~ts -> ~ts",	5✔
290	[LServer, RServer]),	5✔
291	ejabberd_hooks:run_fold(s2s_out_init, ServerHost, {ok, State2}, [Opts]).	5✔
292
293	handle_call(Request, From, #{server_host := ServerHost} = State) ->
294	ejabberd_hooks:run_fold(s2s_out_handle_call, ServerHost, State, [Request, From]).	×
295
296	handle_cast({update_state, Fun}, State) ->
297	case Fun of	×
298	{M, F, A} -> erlang:apply(M, F, [State\|A]);	×
299	_ when is_function(Fun) -> Fun(State)	×
300	end;
301	handle_cast(Msg, #{server_host := ServerHost} = State) ->
302	ejabberd_hooks:run_fold(s2s_out_handle_cast, ServerHost, State, [Msg]).	×
303
304	handle_info({route, Pkt}, #{queue := Q, on_route := Action} = State) ->
305	case Action of	3✔
306	queue ->
307	try State#{queue => p1_queue:in(Pkt, Q)}	3✔
308	catch error:full ->
309	Q1 = p1_queue:set_limit(Q, unlimited),	×
310	Q2 = p1_queue:in(Pkt, Q1),	×
311	handle_stream_end(queue_full, State#{queue => Q2})	×
312	end;
313	bounce -> bounce_packet(Pkt, State);	×
314	send -> set_idle_timeout(send(State, Pkt))	×
315	end;
316	handle_info(Info, #{server_host := ServerHost} = State) ->
317	ejabberd_hooks:run_fold(s2s_out_handle_info, ServerHost, State, [Info]).	×
318
319	terminate(Reason, #{server := LServer,
320	remote_server := RServer} = State) ->
321	State1 = case Reason of	2✔
322	normal -> State;	2✔
323	_ -> State#{stop_reason => internal_failure}	×
324	end,
325	State2 = bounce_queue(State1),	2✔
326	bounce_message_queue({LServer, RServer}, State2).	2✔
327
328	code_change(_OldVsn, State, _Extra) ->
329	{ok, State}.	×
330
331	%%%===================================================================
332	%%% Internal functions
333	%%%===================================================================
334	-spec get_addr_type(inet:ip_address()) -> inet:address_family().
335	get_addr_type({_, _, _, _}) -> inet;	5✔
336	get_addr_type({_, _, _, _, _, _, _, _}) -> inet6.	5✔
337
338	-spec resend_queue(state()) -> state().
339	resend_queue(State) ->
340	queue_fold(	3✔
341	fun(Pkt, AccState) ->
342	send(AccState, Pkt)	3✔
343	end, State).
344
345	-spec bounce_queue(state()) -> state().
346	bounce_queue(State) ->
347	queue_fold(	2✔
348	fun(Pkt, AccState) ->
349	bounce_packet(Pkt, AccState)	×
350	end, State).
351
352	-spec bounce_message_queue({binary(), binary()}, state()) -> state().
353	bounce_message_queue(FromTo, State) ->
354	receive {route, Pkt} ->	2✔
355	State1 = bounce_packet(Pkt, State),	×
356	bounce_message_queue(FromTo, State1)	×
357	after 0 ->
358	State	2✔
359	end.
360
361	-spec bounce_packet(xmpp_element(), state()) -> state().
362	bounce_packet(Pkt, State) when ?is_stanza(Pkt) ->
363	#{server_host := Host} = State,	×
364	case ejabberd_hooks:run_fold(	×
365	s2s_out_bounce_packet, Host, State, [Pkt]) of
366	ignore ->
367	State;	×
368	State2 ->
369	Lang = xmpp:get_lang(Pkt),	×
370	Err = mk_bounce_error(Lang, State2),	×
371	ejabberd_router:route_error(Pkt, Err),	×
372	State2	×
373	end;
374	bounce_packet(_, State) ->
375	State.	×
376
377	-spec mk_bounce_error(binary(), state()) -> stanza_error().
378	mk_bounce_error(Lang, #{stop_reason := Why}) ->
379	Reason = format_error(Why),	×
380	case Why of	×
381	internal_failure ->
382	xmpp:err_internal_server_error(Reason, Lang);	×
383	queue_full ->
384	xmpp:err_resource_constraint(Reason, Lang);	×
385	{dns, _} ->
386	xmpp:err_remote_server_not_found(Reason, Lang);	×
387	{idna, _} ->
388	xmpp:err_remote_server_not_found(Reason, Lang);	×
389	_ ->
390	xmpp:err_remote_server_timeout(Reason, Lang)	×
391	end;
392	mk_bounce_error(_Lang, _State) ->
393	%% We should not be here. Probably :)
394	xmpp:err_remote_server_not_found().	×
395
396	-spec get_delay() -> non_neg_integer().
397	get_delay() ->
398	MaxDelay = ejabberd_option:s2s_max_retry_delay(),	×
399	p1_rand:uniform(MaxDelay).	×
400
401	-spec set_idle_timeout(state()) -> state().
402	set_idle_timeout(#{on_route := send, server_host := ServerHost} = State) ->
403	Timeout = ejabberd_s2s:get_idle_timeout(ServerHost),	3✔
404	xmpp_stream_out:set_timeout(State, Timeout);	3✔
405	set_idle_timeout(State) ->
406	State.	×
407
408	-spec queue_fold(fun((xmpp_element(), state()) -> state()), state()) -> state().
409	queue_fold(F, #{queue := Q} = State) ->
410	case p1_queue:out(Q) of	8✔
411	{{value, Pkt}, Q1} ->
412	State1 = F(Pkt, State#{queue => Q1}),	3✔
413	queue_fold(F, State1);	3✔
414	{empty, Q1} ->
415	State#{queue => Q1}	5✔
416	end.
417
418	format_error(internal_failure) ->
419	<<"Internal server error">>;	×
420	format_error(queue_full) ->
421	<<"Stream queue is overloaded">>;	×
422	format_error(Reason) ->
423	xmpp_stream_out:format_error(Reason).	1✔

processone / ejabberd / 538

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous