diff --git a/config/sys-dev.config b/config/sys-dev.config index 6ab23fc..d4f8350 100644 --- a/config/sys-dev.config +++ b/config/sys-dev.config @@ -26,7 +26,15 @@ {port, 443}, {acceptors, 500}, {max_connections, 10240}, - {backlog, 10240} + {backlog, 10240}, + {certfile, "server.crt"}, + {keyfile, "server.key"}, + {limits, [ + {max_packet_size, 16384}, + {socket_active_n, 100}, + %% 单位为秒 + {heartbeat_sec, 15} + ]} ]}, %% 网络带宽, 单位为: kb diff --git a/config/sys-prod.config b/config/sys-prod.config index c75cff7..7b62d8b 100644 --- a/config/sys-prod.config +++ b/config/sys-prod.config @@ -26,7 +26,15 @@ {port, 443}, {acceptors, 500}, {max_connections, 10240}, - {backlog, 10240} + {backlog, 10240}, + {certfile, "server.crt"}, + {keyfile, "server.key"}, + {limits, [ + {max_packet_size, 16384}, + {socket_active_n, 100}, + %% 单位为秒 + {heartbeat_sec, 15} + ]} ]}, %% 网络带宽, 单位为: kb diff --git a/src/sdlan_sup.erl b/src/sdlan_sup.erl index 72ed0b0..512dc75 100644 --- a/src/sdlan_sup.erl +++ b/src/sdlan_sup.erl @@ -90,6 +90,8 @@ init([]) -> modules => ['sdlan_quic_server'] }, + sdlan_ssl_server:child_spec(), + #{ id => sdlan_sync_mysql, start => {sdlan_sync_mysql, start_link, []}, diff --git a/src/ssl/sdlan_ssl_server.erl b/src/ssl/sdlan_ssl_server.erl index 68c9d7f..32bdd95 100644 --- a/src/ssl/sdlan_ssl_server.erl +++ b/src/ssl/sdlan_ssl_server.erl @@ -10,25 +10,37 @@ -author("anlicheng"). %% API --export([start/0]). +-export([start/0, child_spec/0]). %% 启动ssl服务 start() -> + {TransOpts, Limits, Port} = listener_options(), + {ok, Pid} = ranch:start_listener(ssl_server, ranch_ssl, TransOpts, sdlan_ssl_transport, Limits), + logger:debug("[sdlan_ssl_server] the ssl server start at: ~p, pid is: ~p", [Port, Pid]), + {ok, Pid}. + +child_spec() -> + {TransOpts, Limits, _Port} = listener_options(), + ranch:child_spec(ssl_server, ranch_ssl, TransOpts, sdlan_ssl_transport, Limits). + +listener_options() -> {ok, Props} = application:get_env(sdlan, ssl_server), Acceptors = proplists:get_value(acceptors, Props, 50), MaxConnections = proplists:get_value(max_connections, Props, 10240), Backlog = proplists:get_value(backlog, Props, 1024), Port = proplists:get_value(port, Props), + Limits = proplists:get_value(limits, Props, []), PrivDir = code:priv_dir(sdlan), - CertFile = filename:join([PrivDir, "ssl", "server.crt"]), - KeyFile = filename:join([PrivDir, "ssl", "server.key"]), + CertFile = filename:join([PrivDir, "ssl", proplists:get_value(certfile, Props, "server.crt")]), + KeyFile = filename:join([PrivDir, "ssl", proplists:get_value(keyfile, Props, "server.key")]), TransOpts = #{ max_connections => MaxConnections, num_acceptors => Acceptors, shutdown => brutal_kill, socket_opts => [ + binary, {nodelay, true}, {backlog, Backlog}, {port, Port}, @@ -36,5 +48,4 @@ start() -> {keyfile, KeyFile} ] }, - {ok, _} = ranch:start_listener(ssl_server, ranch_ssl, TransOpts, ssl_channel, []), - logger:debug("[iot_app] the ssl server start at: ~p", [Port]). \ No newline at end of file + {TransOpts, Limits, Port}. diff --git a/src/ssl/sdlan_ssl_transport.erl b/src/ssl/sdlan_ssl_transport.erl new file mode 100644 index 0000000..341ee8a --- /dev/null +++ b/src/ssl/sdlan_ssl_transport.erl @@ -0,0 +1,261 @@ +%%%------------------------------------------------------------------- +%%% @author anlicheng +%%% @copyright (C) 2026, +%%% @doc +%%% SSL transport for sdlan sessions. +%%% @end +%%%------------------------------------------------------------------- +-module(sdlan_ssl_transport). +-author("anlicheng"). +-include("sdlan.hrl"). +-include("sdlan_pb.hrl"). + +-behaviour(gen_statem). +-behaviour(ranch_protocol). + +-define(SOCKET_ACTIVE_N, 100). + +%% Ranch protocol callback +-export([start_link/4]). + +%% gen_statem callbacks +-export([init/1, handle_event/4, terminate/3, code_change/4, callback_mode/0]). + +-record(state, { + ref :: ranch:ref(), + socket :: undefined | ssl:sslsocket(), + transport :: module(), + ok_msg :: atom(), + closed_msg :: atom(), + error_msg :: atom(), + passive_msg :: atom(), + + max_packet_size = 16384, + heartbeat_sec = 15, + socket_active_n = ?SOCKET_ACTIVE_N, + + %% 累积器，用于处理协议framing的解析 + buf = <<>>, + session :: sdlan_session:state(), + + frames_recv = 0, + bytes_recv = 0, + + close_reason = undefined +}). + +%%%=================================================================== +%%% Ranch protocol callback +%%%=================================================================== + +start_link(Ref, Socket, Transport, Limits) -> + gen_statem:start_link(?MODULE, [Ref, Socket, Transport, Limits], []). + +%%%=================================================================== +%%% gen_statem callbacks +%%%=================================================================== + +init([Ref, Socket, Transport, Limits]) -> + MaxPacketSize = proplists:get_value(max_packet_size, Limits, 16384), + HeartbeatSec = proplists:get_value(heartbeat_sec, Limits, 15), + SocketActiveN = proplists:get_value(socket_active_n, Limits, proplists:get_value(stream_active_n, Limits, ?SOCKET_ACTIVE_N)), + {OkMsg, ClosedMsg, ErrorMsg} = Transport:messages(), + Session = sdlan_session:new(HeartbeatSec), + {ok, handshaking, #state{ + ref = Ref, + socket = Socket, + transport = Transport, + ok_msg = OkMsg, + closed_msg = ClosedMsg, + error_msg = ErrorMsg, + passive_msg = passive_msg(OkMsg), + max_packet_size = MaxPacketSize, + heartbeat_sec = HeartbeatSec, + socket_active_n = SocketActiveN, + session = Session + }}. + +callback_mode() -> + handle_event_function. + +handle_event(info, {handshake, Ref, Transport, Socket, Timeout}, handshaking, + State = #state{ref = Ref, transport = Transport, max_packet_size = MaxPacketSize, + heartbeat_sec = HeartbeatSec, socket_active_n = SocketActiveN}) -> + case Transport:handshake(Socket, [], Timeout) of + {ok, SslSocket} -> + ok = Transport:setopts(SslSocket, [{mode, binary}, {active, SocketActiveN}]), + WelcomePkt = welcome_packet(MaxPacketSize, HeartbeatSec), + ssl_send(Transport, SslSocket, <>), + logger:debug("[sdlan_ssl_transport] ssl handshake ok, send welcome"), + {next_state, initialized, State#state{socket = SslSocket}}; + {error, Reason} -> + {stop, {ssl_handshake_failed, Reason}, State} + end; + +handle_event(info, {OkMsg, Socket, Data}, _StateName, + State = #state{socket = Socket, ok_msg = OkMsg, buf = Buf, max_packet_size = MaxPacketSize, + bytes_recv = BytesRecv, frames_recv = FramesRecv}) when is_binary(Data) -> + case decode_frames(<>, MaxPacketSize) of + {error, Reason} -> + {stop, Reason, State}; + {ok, NBuf, Frames} -> + Actions = [{next_event, internal, {frame, Frame}} || Frame <- Frames], + {keep_state, State#state{buf = NBuf, bytes_recv = BytesRecv + byte_size(Data), frames_recv = FramesRecv + length(Frames)}, Actions} + end; + +handle_event(info, {PassiveMsg, Socket}, _StateName, + State = #state{socket = Socket, transport = Transport, passive_msg = PassiveMsg, socket_active_n = SocketActiveN}) -> + ok = Transport:setopts(Socket, [{active, SocketActiveN}]), + {keep_state, State}; + +handle_event(info, {ClosedMsg, Socket}, _StateName, State = #state{socket = Socket, closed_msg = ClosedMsg}) -> + expected_stop(socket_closed, State); + +handle_event(info, {ErrorMsg, Socket, Reason}, _StateName, State = #state{socket = Socket, error_msg = ErrorMsg}) -> + expected_stop({socket_error, Reason}, State); + +%% 处理内部的包消息 +handle_event(internal, {frame, Frame}, StateName, State = #state{socket = Socket, transport = Transport, session = Session}) -> + case sdlan_session:handle_frame(Frame, Session) of + {ok, NSession, Packets} -> + send_packets(Transport, Socket, Packets), + {next_state, next_state_name(StateName, NSession), State#state{session = NSession}}; + {stop, Reason, NSession, Packets} -> + send_packets(Transport, Socket, Packets), + {stop, Reason, State#state{session = NSession}} + end; + +handle_event(info, {timeout, TimerRef, ping_ticker}, _StateName, State = #state{session = Session}) -> + case sdlan_session:handle_timeout(TimerRef, Session) of + {ok, NSession} -> + {next_state, next_state_name(registered, NSession), State#state{session = NSession}}; + {stop, Reason, NSession} -> + expected_stop(Reason, State#state{session = NSession}) + end; + +%% 发送事件信息 +handle_event(cast, {send_event, Event}, _StateName, State = #state{socket = Socket, transport = Transport, session = Session}) -> + case sdlan_session:send_event(Event, Session) of + {ok, NSession, Packets} -> + send_packets(Transport, Socket, Packets), + {keep_state, State#state{session = NSession}}; + {error, not_registered} -> + keep_state_and_data + end; + +%% 发送命令信息 +handle_event(cast, {command, Ref, ReceiverPid, SubCommand}, _StateName, State = #state{socket = Socket, transport = Transport, session = Session}) -> + case sdlan_session:command(Ref, ReceiverPid, SubCommand, Session) of + {ok, NSession, Packets} -> + send_packets(Transport, Socket, Packets), + {keep_state, State#state{session = NSession}}; + {error, not_registered} -> + keep_state_and_data + end; + +handle_event({call, From}, debug_info, StateName, State) -> + {keep_state, State, [{reply, From, debug_info(StateName, State)}]}; + +handle_event(EventType, Info, StateName, State) -> + logger:notice("[sdlan_ssl_transport] state: ~p, state_name: ~p, event_type: ~p, info: ~p", [State, StateName, EventType, Info]), + keep_state_and_data. + +terminate(Reason, _StateName, #state{socket = Socket, transport = Transport, session = Session, close_reason = CloseReason}) -> + Socket =/= undefined andalso catch Transport:close(Socket), + logger:notice("[sdlan_ssl_transport] terminate closed with reason: ~p, close_reason: ~p", [Reason, CloseReason]), + sdlan_session:close(Session), + ok. + +code_change(_OldVsn, StateName, State = #state{}, _Extra) -> + {ok, StateName, State}. + +%%%=================================================================== +%%% Internal functions +%%%=================================================================== + +welcome_packet(MaxPacketSize, HeartbeatSec) -> + Ipv6Assist = case application:get_env(sdlan, ipv6_assist_info) of + {ok, {V6Bytes, Port}} -> + #'SDLV6Info' { + v6 = V6Bytes, + port = Port + }; + _ -> + undefined + end, + sdlan_pb:encode_msg(#'SDLWelcome'{ + version = 1, + max_bidi_streams = 1, + max_packet_size = MaxPacketSize, + heartbeat_sec = HeartbeatSec, + ipv6_assist = Ipv6Assist + }). + +%% 有2种情况 +%% 1. 收到了多个完整的请求 +%% 2. 不完整，则不处理 +-spec decode_frames(Buf :: binary(), MaxPacketSize :: integer()) -> {ok, RestBin::binary(), Frames :: list()} | {error, Reason :: any()}. +decode_frames(Buf, MaxPacketSize) when is_binary(Buf) -> + decode_frames0(Buf, MaxPacketSize, []). +decode_frames0(<>, MaxPacketSize, _Frames) when Len > MaxPacketSize -> + {error, frame_too_large}; +decode_frames0(<>, MaxPacketSize, Frames) -> + decode_frames0(Rest, MaxPacketSize, [Frame|Frames]); +decode_frames0(Rest, _MaxPacketSize, Frames) -> + {ok, Rest, lists:reverse(Frames)}. + +ssl_send(Transport, Socket, Packet) when is_binary(Packet) -> + Len = byte_size(Packet), + true = Len =< 65535, + case Transport:send(Socket, <>) of + ok -> + incr_counter(ssl_frames_sent, 1), + incr_counter(ssl_bytes_sent, Len + 2), + ok; + {error, Reason} -> + exit({ssl_send_failed, Reason}) + end. + +send_packets(Transport, Socket, Packets) -> + lists:foreach(fun(Packet) -> ssl_send(Transport, Socket, Packet) end, Packets). + +expected_stop(Reason, State) -> + logger:notice("[sdlan_ssl_transport] expected close: ~p", [Reason]), + {stop, normal, State#state{close_reason = Reason}}. + +next_state_name(_StateName, Session) -> + sdlan_session:state_name(Session). + +passive_msg(OkMsg) -> + list_to_atom(atom_to_list(OkMsg) ++ "_passive"). + +debug_info(StateName, #state{ + session = Session, + frames_recv = FramesRecv, + bytes_recv = BytesRecv, + socket_active_n = SocketActiveN, + heartbeat_sec = HeartbeatSec +}) -> + ProcInfo = maps:from_list(process_info(self(), [message_queue_len, memory, reductions])), + SessionInfo = sdlan_session:debug_info(Session), + maps:merge(SessionInfo, ProcInfo#{ + state => StateName, + session => SessionInfo, + frames_recv => FramesRecv, + frames_sent => get_counter(ssl_frames_sent), + bytes_recv => BytesRecv, + bytes_sent => get_counter(ssl_bytes_sent), + socket_active_n => SocketActiveN, + heartbeat_sec => HeartbeatSec + }). + +incr_counter(Key, Inc) -> + erlang:put(Key, get_counter(Key) + Inc). + +get_counter(Key) -> + case erlang:get(Key) of + undefined -> + 0; + Value when is_integer(Value) -> + Value + end.