aboutsummaryrefslogtreecommitdiffstats
path: root/lib/diameter/src
diff options
context:
space:
mode:
Diffstat (limited to 'lib/diameter/src')
-rw-r--r--lib/diameter/src/base/diameter.erl3
-rw-r--r--lib/diameter/src/base/diameter_peer_fsm.erl90
-rw-r--r--lib/diameter/src/base/diameter_traffic.erl32
-rw-r--r--lib/diameter/src/transport/diameter_sctp.erl7
-rw-r--r--lib/diameter/src/transport/diameter_tcp.erl119
5 files changed, 153 insertions, 98 deletions
diff --git a/lib/diameter/src/base/diameter.erl b/lib/diameter/src/base/diameter.erl
index 6be544e950..f563d244f6 100644
--- a/lib/diameter/src/base/diameter.erl
+++ b/lib/diameter/src/base/diameter.erl
@@ -332,8 +332,9 @@ call(SvcName, App, Message) ->
| {capabilities_cb, evaluable()}
| {capx_timeout, 'Unsigned32'()}
| {disconnect_cb, evaluable()}
- | {watchdog_timer, 'Unsigned32'() | {module(), atom(), list()}}
+ | {length_errors, exit | handle | discard}
| {reconnect_timer, 'Unsigned32'()}
+ | {watchdog_timer, 'Unsigned32'() | {module(), atom(), list()}}
| {private, any()}.
%% Predicate passed to remove_transport/2
diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl
index ad26f230ef..66342f7b62 100644
--- a/lib/diameter/src/base/diameter_peer_fsm.erl
+++ b/lib/diameter/src/base/diameter_peer_fsm.erl
@@ -18,10 +18,10 @@
%%
%%
-%% This module implements (as a process) the RFC 3588 Peer State
+%% This module implements (as a process) the RFC 3588/6733 Peer State
%% Machine modulo the necessity of adapting the peer election to the
-%% fact that we don't know the identity of a peer until we've
-%% received a CER/CEA from it.
+%% fact that we don't know the identity of a peer until we've received
+%% a CER/CEA from it.
%%
-module(diameter_peer_fsm).
@@ -107,8 +107,9 @@
transport :: pid(), %% transport process
dictionary :: module(), %% common dictionary
service :: #diameter_service{},
- dpr = false :: false | {uint32(), uint32()}}).
+ dpr = false :: false | {uint32(), uint32()},
%% | hop by hop and end to end identifiers
+ length_errors :: exit | handle | discard}).
%% There are non-3588 states possible as a consequence of 5.6.1 of the
%% standard and the corresponding problem for incoming CEA's: we don't
@@ -191,15 +192,22 @@ i({Ack, WPid, {M, Ref} = T, Opts, {Mask,
putr(?REF_KEY, Ref),
putr(?SEQUENCE_KEY, Mask),
putr(?RESTRICT_KEY, Nodes),
- {TPid, Addrs} = start_transport(T, Rest, Svc),
+
Tmo = proplists:get_value(capx_timeout, Opts, ?EVENT_TIMEOUT),
?IS_TIMEOUT(Tmo) orelse ?ERROR({invalid, {capx_timeout, Tmo}}),
+ OnLengthErr = proplists:get_value(length_errors, Opts, exit),
+ lists:member(OnLengthErr, [exit, handle, discard])
+ orelse ?ERROR({invalid, {length_errors, OnLengthErr}}),
+
+ {TPid, Addrs} = start_transport(T, Rest, Svc),
+
#state{state = {'Wait-Conn-Ack', Tmo},
parent = WPid,
transport = TPid,
dictionary = Dict0,
mode = M,
- service = svc(Svc, Addrs)}.
+ service = svc(Svc, Addrs),
+ length_errors = OnLengthErr}.
%% The transport returns its local ip addresses so that different
%% transports on the same service can use different local addresses.
%% The local addresses are put into Host-IP-Address avps here when
@@ -512,21 +520,6 @@ encode(Rec, Dict) ->
%% recv/2
-%% RFC 3588 has result code 5015 for an invalid length but if a
-%% transport is detecting message boundaries using the length header
-%% then a length error will likely lead to further errors.
-
-recv(#diameter_packet{header = #diameter_header{length = Len}
- = Hdr,
- bin = Bin},
- S)
- when Len < 20;
- (0 /= Len rem 4 orelse bit_size(Bin) /= 8*Len) ->
- discard(invalid_message_length, recv, [size(Bin),
- bit_size(Bin) rem 8,
- Hdr,
- S]);
-
recv(#diameter_packet{header = #diameter_header{} = Hdr}
= Pkt,
#state{parent = Pid,
@@ -541,29 +534,52 @@ recv(#diameter_packet{header = undefined,
bin = Bin}
= Pkt,
S) ->
- recv(Pkt#diameter_packet{header = diameter_codec:decode_header(Bin)}, S);
+ recv(diameter_codec:decode_header(Bin), Pkt, S);
-recv(Bin, S)
- when is_binary(Bin) ->
- recv(#diameter_packet{bin = Bin}, S);
+recv(Bin, S) ->
+ recv(#diameter_packet{bin = Bin}, S).
-recv(#diameter_packet{header = false} = Pkt, S) ->
- discard(truncated_header, recv, [Pkt, S]).
+%% recv/3
-msg_id({_,_,_} = T, _) ->
- T;
-msg_id(_, Hdr) ->
- diameter_codec:msg_id(Hdr).
+recv(#diameter_header{length = Len}
+ = H,
+ #diameter_packet{bin = Bin}
+ = Pkt,
+ #state{length_errors = E}
+ = S)
+ when E == handle;
+ 0 == Len rem 4, bit_size(Bin) == 8*Len ->
+ recv(Pkt#diameter_packet{header = H}, S);
+
+recv(#diameter_header{}
+ = H,
+ #diameter_packet{bin = Bin},
+ #state{length_errors = E}
+ = S) ->
+ invalid(E,
+ invalid_message_length,
+ recv,
+ [size(Bin), bit_size(Bin) rem 8, H, S]);
-%% Treat invalid length as a transport error and die. Especially in
-%% the TCP case, in which there's no telling where the next message
-%% begins in the incoming byte stream, keeping a crippled connection
-%% alive may just make things worse.
+recv(false, Pkt, #state{length_errors = E} = S) ->
+ invalid(E, truncated_header, recv, [Pkt, S]).
-discard(Reason, F, A) ->
+%% Note that counters here only count discarded messages.
+invalid(E, Reason, F, A) ->
diameter_stats:incr(Reason),
+ abort(E, Reason, F, A).
+
+abort(exit, Reason, F, A) ->
diameter_lib:warning_report(Reason, {?MODULE, F, A}),
- throw({?MODULE, abort, Reason}).
+ throw({?MODULE, abort, Reason});
+
+abort(_, _, _, _) ->
+ ok.
+
+msg_id({_,_,_} = T, _) ->
+ T;
+msg_id(_, Hdr) ->
+ {_,_,_} = diameter_codec:msg_id(Hdr).
%% rcv/3
diff --git a/lib/diameter/src/base/diameter_traffic.erl b/lib/diameter/src/base/diameter_traffic.erl
index 2f486861a2..0de3825943 100644
--- a/lib/diameter/src/base/diameter_traffic.erl
+++ b/lib/diameter/src/base/diameter_traffic.erl
@@ -309,21 +309,35 @@ request_cb(App,
%% examine/1
%%
-%% Look for errors in a decoded message. Length errors result in
-%% decode failure in diameter_codec.
+%% Look for errors in a decoded message. It's odd/unfortunate that
+%% 501[15] aren't protocol errors.
-examine(#diameter_packet{header = #diameter_header{version
- = ?DIAMETER_VERSION}}
- = Pkt) ->
- Pkt;
+%% DIAMETER_INVALID_MESSAGE_LENGTH 5015
+%%
+%% This error is returned when a request is received with an invalid
+%% message length.
+
+examine(#diameter_packet{header = #diameter_header{length = Len},
+ bin = Bin,
+ errors = Es}
+ = Pkt)
+ when Len < 20;
+ 0 /= Len rem 4;
+ 8*Len /= bit_size(Bin) ->
+ Pkt#diameter_packet{errors = [5015 | Es]};
%% DIAMETER_UNSUPPORTED_VERSION 5011
%% This error is returned when a request was received, whose version
%% number is unsupported.
-examine(#diameter_packet{errors = Es} = Pkt) ->
- Pkt#diameter_packet{errors = [5011 | Es]}.
-%% It's odd/unfortunate that this isn't a protocol error.
+examine(#diameter_packet{header = #diameter_header{version = V},
+ errors = Es}
+ = Pkt)
+ when V /= ?DIAMETER_VERSION ->
+ Pkt#diameter_packet{errors = [5011 | Es]};
+
+examine(Pkt) ->
+ Pkt.
%% request_cb/8
diff --git a/lib/diameter/src/transport/diameter_sctp.erl b/lib/diameter/src/transport/diameter_sctp.erl
index ac79fde07e..8b8c2a6694 100644
--- a/lib/diameter/src/transport/diameter_sctp.erl
+++ b/lib/diameter/src/transport/diameter_sctp.erl
@@ -484,8 +484,8 @@ transition({diameter, {close, Pid}}, #transport{parent = Pid}) ->
%% TLS over SCTP is described in RFC 3436 but has limitations as
%% described in RFC 6083. The latter describes DTLS over SCTP, which
%% addresses these limitations, DTLS itself being described in RFC
-%% 4347. TLS is primarily used over TCP, which the current RFC 3588
-%% draft acknowledges by equating TLS with TLS/TCP and DTLS/SCTP.
+%% 4347. TLS is primarily used over TCP, which RFC 6733 acknowledges
+%% by equating TLS with TLS/TCP and DTLS/SCTP.
transition({diameter, {tls, _Ref, _Type, _Bool}}, _) ->
stop;
@@ -585,8 +585,7 @@ recv({_, #sctp_assoc_change{state = comm_up,
socket = Sock}
= S) ->
Ref = getr(?REF_KEY),
- is_reference(Ref) %% started in new code
- andalso publish(T, Ref, Id, Sock),
+ publish(T, Ref, Id, Sock),
up(S#transport{assoc_id = Id,
streams = {IS, OS}});
diff --git a/lib/diameter/src/transport/diameter_tcp.erl b/lib/diameter/src/transport/diameter_tcp.erl
index 596e582ab0..132088b514 100644
--- a/lib/diameter/src/transport/diameter_tcp.erl
+++ b/lib/diameter/src/transport/diameter_tcp.erl
@@ -52,7 +52,10 @@
-define(DEFAULT_PORT, 3868). %% RFC 3588, ch 2.1
-define(LISTENER_TIMEOUT, 30000).
--define(FRAGMENT_TIMEOUT, 1000).
+-define(DEFAULT_FRAGMENT_TIMEOUT, 1000).
+
+-define(IS_UINT32(N), (is_integer(N) andalso 0 =< N andalso 0 == N bsr 32)).
+-define(IS_TIMEOUT(N), (infinity == N orelse ?IS_UINT32(N))).
%% cb_info passed to ssl.
-define(TCP_CB(Mod), {Mod, tcp, tcp_closed, tcp_error}).
@@ -72,7 +75,6 @@
{parent :: pid(),
transport = self() :: pid()}).
--type tref() :: reference(). %% timer reference
-type length() :: 0..16#FFFFFF. %% message length from Diameter header
-type size() :: non_neg_integer(). %% accumulated binary size
-type frag() :: {length(), size(), binary(), list(binary())}
@@ -83,8 +85,11 @@
{socket :: inet:socket() | ssl:sslsocket(), %% accept/connect socket
parent :: pid(), %% of process that started us
module :: module(), %% gen_tcp-like module
- frag = <<>> :: binary() | {tref(), frag()}, %% message fragment
- ssl :: boolean() | [term()]}). %% ssl options
+ frag = <<>> :: frag(), %% message fragment
+ ssl :: boolean() | [term()], %% ssl options
+ timeout :: infinity | 0..16#FFFFFFFF, %% fragment timeout
+ tref = false :: false | reference(), %% fragment timer reference
+ flush = false :: boolean()}). %% flush fragment at timeout?
%% The usual transport using gen_tcp can be replaced by anything
%% sufficiently gen_tcp-like by passing a 'module' option as the first
%% (for simplicity) transport option. The transport_module diameter_etcp
@@ -161,7 +166,12 @@ i({T, Ref, Mod, Pid, Opts, Addrs})
%% that does nothing but kill us with the parent until call
%% returns.
{ok, MPid} = diameter_tcp_sup:start_child(#monitor{parent = Pid}),
- {SslOpts, Rest} = ssl(Opts),
+ {SslOpts, Rest0} = ssl(Opts),
+ {OwnOpts, Rest} = own(Rest0),
+ Tmo = proplists:get_value(fragment_timer,
+ OwnOpts,
+ ?DEFAULT_FRAGMENT_TIMEOUT),
+ ?IS_TIMEOUT(Tmo) orelse ?ERROR({fragment_timer, Tmo}),
Sock = i(T, Ref, Mod, Pid, SslOpts, Rest, Addrs),
MPid ! {stop, self()}, %% tell the monitor to die
M = if SslOpts -> ssl; true -> Mod end,
@@ -170,7 +180,8 @@ i({T, Ref, Mod, Pid, Opts, Addrs})
#transport{parent = Pid,
module = M,
socket = Sock,
- ssl = SslOpts};
+ ssl = SslOpts,
+ timeout = Tmo};
%% Put the reference in the process dictionary since we now use it
%% advertise the ssl socket after TLS upgrade.
@@ -196,6 +207,10 @@ i({listen, LRef, APid, {Mod, Opts, Addrs}}) ->
erlang:monitor(process, APid),
start_timer(#listener{socket = LSock}).
+own(Opts) ->
+ {Own, Rest} = proplists:split(Opts, [fragment_timer]),
+ {lists:append(Own), Rest}.
+
ssl(Opts) ->
{[SslOpts], Rest} = proplists:split(Opts, [ssl_options]),
{ssl_opts(SslOpts), Rest}.
@@ -450,6 +465,7 @@ t(T,S) ->
%% Initial incoming message when we might need to upgrade to TLS:
%% don't request another message until we know.
+
transition({tcp, Sock, Bin}, #transport{socket = Sock,
parent = Pid,
frag = Head,
@@ -457,13 +473,13 @@ transition({tcp, Sock, Bin}, #transport{socket = Sock,
ssl = Opts}
= S)
when is_list(Opts) ->
- case recv1(Head, Bin) of
+ case rcv(Head, Bin) of
{Msg, B} when is_binary(Msg) ->
diameter_peer:recv(Pid, Msg),
S#transport{frag = B};
Frag ->
setopts(M, Sock),
- S#transport{frag = Frag}
+ start_fragment_timer(S#transport{frag = Frag})
end;
%% Incoming message.
@@ -474,7 +490,7 @@ transition({P, Sock, Bin}, #transport{socket = Sock,
when P == tcp, not B;
P == ssl, B ->
setopts(M, Sock),
- recv(Bin, S);
+ start_fragment_timer(recv(Bin, S));
%% Capabilties exchange has decided on whether or not to run over TLS.
transition({diameter, {tls, Ref, Type, B}}, #transport{parent = Pid}
@@ -485,7 +501,7 @@ transition({diameter, {tls, Ref, Type, B}}, #transport{parent = Pid}
= tls_handshake(Type, B, S),
Pid ! {diameter, {tls, Ref}},
setopts(M, Sock),
- NS#transport{ssl = B};
+ start_fragment_timer(NS#transport{ssl = B});
transition({C, Sock}, #transport{socket = Sock,
ssl = B})
@@ -518,8 +534,8 @@ transition({diameter, {close, Pid}}, #transport{parent = Pid,
stop;
%% Timeout for reception of outstanding packets.
-transition({timeout, TRef, flush}, S) ->
- flush(TRef, S);
+transition({timeout, TRef, flush}, #transport{tref = TRef} = S) ->
+ flush(S#transport{tref = false});
%% Request for the local port number.
transition({resolve_port, Pid}, #transport{socket = Sock,
@@ -557,9 +573,7 @@ tls_handshake(Type, true, #transport{socket = Sock,
= S) ->
{ok, SSock} = tls(Type, Sock, [{cb_info, ?TCP_CB(M)} | Opts]),
Ref = getr(?REF_KEY),
- is_reference(Ref) %% started in new code
- andalso
- (true = diameter_reg:add_new({?MODULE, Type, {Ref, SSock}})),
+ true = diameter_reg:add_new({?MODULE, Type, {Ref, SSock}}),
S#transport{socket = SSock,
module = ssl};
@@ -574,30 +588,25 @@ tls(accept, Sock, Opts) ->
%% recv/2
%%
-%% Reassemble fragmented messages and extract multple message sent
+%% Reassemble fragmented messages and extract multiple message sent
%% using Nagle.
recv(Bin, #transport{parent = Pid, frag = Head} = S) ->
- case recv1(Head, Bin) of
+ case rcv(Head, Bin) of
{Msg, B} when is_binary(Msg) ->
diameter_peer:recv(Pid, Msg),
recv(B, S#transport{frag = <<>>});
Frag ->
- S#transport{frag = Frag}
+ S#transport{frag = Frag,
+ flush = false}
end.
-%% recv1/2
+%% rcv/2
%% No previous fragment.
-recv1(<<>>, Bin) ->
+rcv(<<>>, Bin) ->
rcv(Bin);
-recv1({TRef, Head}, Bin) ->
- erlang:cancel_timer(TRef),
- rcv(Head, Bin).
-
-%% rcv/2
-
%% Not even the first four bytes of the header.
rcv(Head, Bin)
when is_binary(Head) ->
@@ -612,22 +621,22 @@ rcv({Len, N, Head, Acc}, Bin) ->
%% Extract a message for which we have all bytes.
rcv(Len, N, Head, Acc)
when Len =< N ->
- rcv1(Len, bin(Head, Acc));
+ recv1(Len, bin(Head, Acc));
%% Wait for more packets.
rcv(Len, N, Head, Acc) ->
- {start_timer(), {Len, N, Head, Acc}}.
+ {Len, N, Head, Acc}.
-%% rcv/2
+%% rcv/1
%% Nothing left.
rcv(<<>> = Bin) ->
Bin;
-%% Well, this isn't good. Chances are things will go south from here
-%% but if we're lucky then the bytes we have extend to an intended
-%% message boundary and we can recover by simply discarding them,
-%% which is the result of receiving them.
+%% The Message Length isn't even sufficient for a header. Chances are
+%% things will go south from here but if we're lucky then the bytes we
+%% have extend to an intended message boundary and we can recover by
+%% simply receiving them. Make it so.
rcv(<<_:1/binary, Len:24, _/binary>> = Bin)
when Len < 20 ->
{Bin, <<>>};
@@ -635,23 +644,23 @@ rcv(<<_:1/binary, Len:24, _/binary>> = Bin)
%% Enough bytes to extract a message.
rcv(<<_:1/binary, Len:24, _/binary>> = Bin)
when Len =< size(Bin) ->
- rcv1(Len, Bin);
+ recv1(Len, Bin);
%% Or not: wait for more packets.
rcv(<<_:1/binary, Len:24, _/binary>> = Head) ->
- {start_timer(), {Len, size(Head), Head, []}};
+ {Len, size(Head), Head, []};
%% Not even 4 bytes yet.
rcv(Head) ->
- {start_timer(), Head}.
+ Head.
-%% rcv1/2
+%% recv1/2
-rcv1(Len, Bin) ->
+recv1(Len, Bin) ->
<<Msg:Len/binary, Rest/binary>> = Bin,
{Msg, Rest}.
-%% bin/[12]
+%% bin/1-2
bin(Head, Acc) ->
list_to_binary([Head | lists:reverse(Acc)]).
@@ -662,7 +671,7 @@ bin(Bin)
when is_binary(Bin) ->
Bin.
-%% start_timer/0
+%% flush/1
%% An erroneously large message length may leave us with a fragment
%% that lingers if the peer doesn't have anything more to send. Start
@@ -675,14 +684,30 @@ bin(Bin)
%% since all messages with length problems are discarded this should
%% also eventually lead to watchdog failover.
-start_timer() ->
- erlang:start_timer(?FRAGMENT_TIMEOUT, self(), flush).
+%% No fragment to flush.
+flush(#transport{frag = <<>>} = S) ->
+ S;
-flush(TRef, #transport{parent = Pid, frag = {TRef, Head}} = S) ->
- diameter_peer:recv(Pid, bin(Head)),
- S#transport{frag = <<>>};
-flush(_, S) ->
- S.
+%% Messages have been received since last timer expiry.
+flush(#transport{flush = false} = S) ->
+ start_fragment_timer(S#transport{flush = true});
+
+%% No messages since last expiry.
+flush(#transport{frag = Frag, parent = Pid} = S) ->
+ diameter_peer:recv(Pid, bin(Frag)),
+ S#transport{frag = <<>>}.
+
+%% start_fragment_timer/1
+%%
+%% Start a timer only if there's none running and a message to flush.
+
+start_fragment_timer(#transport{frag = B, tref = TRef} = S)
+ when B == <<>>;
+ TRef /= false ->
+ S;
+
+start_fragment_timer(#transport{timeout = Tmo} = S) ->
+ S#transport{tref = erlang:start_timer(Tmo, self(), flush)}.
%% accept/2
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-15 16:28:01 +0000