%%% -*-mode:erlang;coding:utf-8;tab-width:4;c-basic-offset:4;indent-tabs-mode:()-*- %%% ex: set ft=erlang fenc=utf-8 sts=4 ts=4 sw=4 et: %%% %%% Copyright 2015 Panagiotis Papadomitsos. All Rights Reserved. %%% Copyright 2021 Miniclip. All Rights Reserved. %%% %%% Original concept inspired and some code copied from %%% https://erlangcentral.org/wiki/index.php?title=Building_a_Non-blocking_TCP_server_using_OTP_principles -module(gen_rpc_acceptor). -author("Panagiotis Papadomitsos "). %%% Behaviour -behaviour(gen_statem). %%% Include the HUT library -include_lib("hut/include/hut.hrl"). %%% Include this library's name macro -include("app.hrl"). %%% Local state -record(state, {socket = undefined :: port() | undefined, driver :: atom(), driver_mod :: atom(), driver_closed :: atom(), driver_error :: atom(), peer :: {inet:ip4_address(), inet:port_number()}, control :: whitelist | blacklist | disabled, list :: sets:set() | undefined}). -elvis([{elvis_style, state_record_and_type, disable}]). %%% Ignore dialyzer warning for call_middleman %%% The non-local return is deliberate -dialyzer([{no_return, [call_middleman/3]}]). %%% Server functions -export([start_link/2, set_socket/2, stop/1]). %% gen_statem callbacks -export([init/1, handle_event/4, callback_mode/0, terminate/3, code_change/4]). %% State machine states -export([waiting_for_socket/3, waiting_for_auth/3, waiting_for_data/3]). %%% Process exports -export([call_worker/6, call_middleman/3]). -ignore_xref(waiting_for_data/3). -ignore_xref(call_middleman/3). -ignore_xref(start_link/2). -ignore_xref(stop/1). -ignore_xref(waiting_for_socket/3). -ignore_xref(waiting_for_auth/3). %%% =================================================== %%% Supervisor functions %%% =================================================== -spec start_link(atom(), {inet:ip4_address(), inet:port_number()}) -> gen_statem:start_ret(). start_link(Driver, Peer) when is_atom(Driver), is_tuple(Peer) -> Name = gen_rpc_helper:make_process_name("acceptor", Peer), gen_statem:start_link({local,Name}, ?MODULE, {Driver, Peer}, []). -spec stop(pid()) -> ok. stop(Pid) when is_pid(Pid) -> gen_statem:stop(Pid, normal, infinity). %%% =================================================== %%% Server functions %%% =================================================== -spec set_socket(pid(), gen_tcp:socket()) -> ok. set_socket(Pid, Socket) when is_pid(Pid) -> gen_statem:call(Pid, {socket_ready,Socket}, infinity). %%% =================================================== %%% Behaviour callbacks %%% =================================================== init({Driver, Peer}) -> ok = gen_rpc_helper:set_optimal_process_flags(), {Control, ControlList} = gen_rpc_helper:get_rpc_module_control(), {DriverMod, _DriverPort, DriverClosed, DriverError} = gen_rpc_helper:get_server_driver_options(Driver), ?log(info, "event=start driver=~s peer=\"~s\"", [Driver, gen_rpc_helper:peer_to_string(Peer)]), {ok, waiting_for_socket, #state{driver=Driver, driver_mod=DriverMod, driver_error=DriverError, driver_closed=DriverClosed, peer=Peer, control=Control, list=ControlList}}. callback_mode() -> state_functions. waiting_for_socket({call,From}, {socket_ready,Socket}, #state{driver=Driver, driver_mod=DriverMod, peer=Peer} = State) -> % Now we own the socket ?log(debug, "event=acquiring_socket_ownership driver=~s socket=\"~s\" peer=\"~p\"", [Driver, gen_rpc_helper:socket_to_string(Socket), gen_rpc_helper:peer_to_string(Peer)]), ok = DriverMod:set_acceptor_opts(Socket), ok = DriverMod:activate_socket(Socket), ok = gen_statem:reply(From, ok), {next_state, waiting_for_auth, State#state{socket=Socket}, gen_rpc_helper:get_authentication_timeout()}. waiting_for_auth(info, {Driver,Socket,Data}, #state{socket=Socket, driver=Driver, driver_mod=DriverMod, peer=Peer} = State) -> case DriverMod:authenticate_client(Socket, Peer, Data) of {error, Reason} -> {stop, Reason, State}; ok -> {next_state, waiting_for_data, State} end; waiting_for_auth(timeout, _Timeout, #state{socket=Socket, driver=Driver, peer=Peer} = State) -> ?log(notice, "event=timed_out_waiting_for_auth driver=~s socket=\"~s\" peer=\"~s\"", [Driver, gen_rpc_helper:socket_to_string(Socket), gen_rpc_helper:peer_to_string(Peer)]), {stop, timed_out_waiting_for_auth, State}; waiting_for_auth(info, {DriverClosed, Socket} = Msg, #state{socket=Socket, driver_closed=DriverClosed} = State) -> handle_event(info, Msg, waiting_for_auth, State); waiting_for_auth(info, {DriverError, Socket, _Reason} = Msg, #state{socket=Socket, driver_error=DriverError} = State) -> handle_event(info, Msg, waiting_for_auth, State). waiting_for_data(info, {Driver,Socket,Data}, #state{socket=Socket, driver=Driver, driver_mod=DriverMod, peer=Peer, control=Control, list=List} = State) -> %% The meat of the whole project: process a function call and return %% the data try erlang:binary_to_term(Data) of {{CallType,M,F,A}, Caller} when CallType =:= call; CallType =:= async_call -> {ModVsnAllowed, RealM} = check_module_version_compat(M), case check_if_module_allowed(RealM, Control, List) of true -> case ModVsnAllowed of true -> WorkerPid = erlang:spawn(?MODULE, call_worker, [self(), CallType, RealM, F, A, Caller]), ?log(debug, "event=call_received driver=~s socket=\"~s\" peer=\"~s\" caller=\"~p\" worker_pid=\"~p\"", [Driver, gen_rpc_helper:socket_to_string(Socket), gen_rpc_helper:peer_to_string(Peer), Caller, WorkerPid]), ok = DriverMod:activate_socket(Socket), {keep_state_and_data, gen_rpc_helper:get_inactivity_timeout(?MODULE)}; false -> ?log(debug, "event=incompatible_module_version driver=~s socket=\"~s\" method=~s module=~s", [Driver, gen_rpc_helper:socket_to_string(Socket), CallType, RealM]), ok = DriverMod:activate_socket(Socket), waiting_for_data(info, {CallType, Caller, {badrpc,incompatible}}, State) end; false -> ?log(debug, "event=request_not_allowed driver=~s socket=\"~s\" control=~s method=~s module=~s", [Driver, gen_rpc_helper:socket_to_string(Socket), Control, CallType, RealM]), ok = DriverMod:activate_socket(Socket), waiting_for_data(info, {CallType, Caller, {badrpc,unauthorized}}, State) end; {cast, M, F, A} -> {ModVsnAllowed, RealM} = check_module_version_compat(M), _Result = case check_if_module_allowed(RealM, Control, List) of true -> case ModVsnAllowed of true -> ?log(debug, "event=cast_received driver=~s socket=\"~s\" peer=\"~s\" module=~s function=~s args=\"~p\"", [Driver, gen_rpc_helper:socket_to_string(Socket), gen_rpc_helper:peer_to_string(Peer), RealM, F, A]), _Pid = erlang:spawn(RealM, F, A); false -> ?log(debug, "event=incompatible_module_version driver=~s socket=\"~s\" module=~s", [Driver, gen_rpc_helper:socket_to_string(Socket), RealM]) end; false -> ?log(debug, "event=request_not_allowed driver=~s socket=\"~s\" control=~s method=cast module=~s", [Driver, gen_rpc_helper:socket_to_string(Socket), Control, RealM]) end, ok = DriverMod:activate_socket(Socket), {keep_state_and_data, gen_rpc_helper:get_inactivity_timeout(?MODULE)}; {abcast, Name, Msg} -> _Result = case check_if_module_allowed(erlang, Control, List) of true -> ?log(debug, "event=abcast_received driver=~s socket=\"~s\" peer=\"~s\" process=~s message=\"~p\"", [Driver, gen_rpc_helper:socket_to_string(Socket), gen_rpc_helper:peer_to_string(Peer), Name, Msg]), Msg = erlang:send(Name, Msg); false -> ?log(debug, "event=request_not_allowed driver=~s socket=\"~s\" control=~s method=~s", [Driver, gen_rpc_helper:socket_to_string(Socket), Control, abcast]) end, ok = DriverMod:activate_socket(Socket), {keep_state_and_data, gen_rpc_helper:get_inactivity_timeout(?MODULE)}; {sbcast, Name, Msg, Caller} -> Reply = case check_if_module_allowed(erlang, Control, List) of true -> ?log(debug, "event=sbcast_received driver=~s socket=\"~s\" peer=\"~s\" process=~s message=\"~p\"", [Driver, gen_rpc_helper:socket_to_string(Socket), gen_rpc_helper:peer_to_string(Peer), Name, Msg]), case erlang:whereis(Name) of undefined -> error; Pid -> Msg = erlang:send(Pid, Msg), success end; false -> ?log(debug, "event=request_not_allowed driver=~s socket=\"~s\" control=~s method=~s", [Driver, gen_rpc_helper:socket_to_string(Socket), Control, sbcast]), error end, ok = DriverMod:activate_socket(Socket), waiting_for_data(info, {sbcast, Caller, Reply}, State); OtherData -> ?log(debug, "event=erroneous_data_received driver=~s socket=\"~s\" peer=\"~s\" data=\"~p\"", [Driver, gen_rpc_helper:socket_to_string(Socket), gen_rpc_helper:peer_to_string(Peer), OtherData]), {stop, {badrpc,erroneous_data}, State} catch error:badarg -> {stop, {badtcp,corrupt_data}, State} end; %% Handle a call worker message waiting_for_data(info, {CallReply,_Caller,_Reply} = Payload, #state{socket=Socket, driver=Driver, driver_mod=DriverMod} = State) when CallReply =:= call orelse CallReply =:= async_call orelse CallReply =:= sbcast -> Packet = erlang:term_to_binary(Payload), ?log(debug, "message=call_reply event=call_reply_received driver=~s socket=\"~s\" type=~s", [Driver, gen_rpc_helper:socket_to_string(Socket), CallReply]), case DriverMod:send(Socket, Packet) of ok -> ?log(debug, "message=call_reply event=call_reply_sent driver=~s socket=\"~s\"", [Driver, gen_rpc_helper:socket_to_string(Socket)]), {keep_state_and_data, gen_rpc_helper:get_inactivity_timeout(?MODULE)}; {error, Reason} -> ?log(error, "message=call_reply event=failed_to_send_call_reply driver=~s socket=\"~s\" reason=\"~p\"", [Driver, gen_rpc_helper:socket_to_string(Socket), Reason]), {stop, Reason, State} end; %% Handle the inactivity timeout gracefully waiting_for_data(timeout, _Undefined, #state{socket=Socket, driver=Driver} = State) -> ?log(info, "message=timeout event=server_inactivity_timeout driver=~s socket=\"~s\" action=stopping", [Driver, gen_rpc_helper:socket_to_string(Socket)]), {stop, normal, State}; waiting_for_data(info, {DriverClosed, Socket} = Msg, #state{socket=Socket, driver_closed=DriverClosed} = State) -> handle_event(info, Msg, waiting_for_data, State); waiting_for_data(info, {DriverError, Socket, _Reason} = Msg, #state{socket=Socket, driver_error=DriverError} = State) -> handle_event(info, Msg, waiting_for_data, State). handle_event(info, {DriverClosed, Socket}, _StateName, #state{socket=Socket, driver=Driver, driver_closed=DriverClosed, peer=Peer} = State) -> ?log(notice, "message=channel_closed driver=~s socket=\"~s\" peer=\"~s\" action=stopping", [Driver, gen_rpc_helper:socket_to_string(Socket), gen_rpc_helper:peer_to_string(Peer)]), {stop, normal, State}; handle_event(info, {DriverError, Socket, Reason}, _StateName, #state{socket=Socket, driver=Driver, driver_error=DriverError, peer=Peer} = State) -> ?log(error, "message=channel_error driver=~s socket=\"~s\" peer=\"~s\" reason=\"~p\" action=stopping", [Driver, gen_rpc_helper:socket_to_string(Socket), gen_rpc_helper:peer_to_string(Peer), Reason]), {stop, normal, State}; handle_event(EventType, Event, StateName, #state{socket=Socket, driver=Driver} = State) -> ?log(error, "event=uknown_event driver=~s socket=\"~s\" event_type=\"~p\" payload=\"~p\" action=stopping", [Driver, gen_rpc_helper:socket_to_string(Socket), EventType, Event]), {stop, {StateName, undefined_event, Event}, State}. terminate(_Reason, _StateName, _State) -> ok. code_change(_OldVsn, StateName, State, _Extra) -> {ok, StateName, State}. %%% =================================================== %%% Private functions %%% =================================================== %% Process an RPC call request outside of the state machine call_worker(Server, CallType, M, F, A, Caller) -> ?log(debug, "event=call_received caller=\"~p\" module=~s function=~s args=\"~p\"", [Caller, M, F, A]), % If called MFA return exception, not of type term(). % This fails term_to_binary coversion, crashes process % and manifest as timeout. Wrap inside anonymous function with catch % will crash the worker quickly not manifest as a timeout. % See call_MFA_undef test. {MPid, MRef} = erlang:spawn_monitor(?MODULE, call_middleman, [M,F,A]), receive {'DOWN', MRef, process, MPid, {call_middleman_result, Res}} -> Server ! {CallType, Caller, Res}; {'DOWN', MRef, process, MPid, AbnormalExit} -> Server ! {CallType, Caller, {badrpc, AbnormalExit}} end. call_middleman(M, F, A) -> Res = try erlang:apply(M, F, A) catch throw:Term -> Term; exit:Reason -> {badrpc, {'EXIT', Reason}}; error:Reason:Stacktrace -> {badrpc, {'EXIT', {Reason, Stacktrace}}} end, erlang:exit({call_middleman_result, Res}), ok. %% Check if the function is RPC-enabled check_if_module_allowed(_Module, disabled, _List) -> true; check_if_module_allowed(Module, whitelist, List) -> sets:is_element(Module, List); check_if_module_allowed(Module, blacklist, List) -> not sets:is_element(Module, List). %% Check if the module version called is compatible with the one %% requested by the caller check_module_version_compat({M, Version}) -> try Attrs = M:module_info(attributes), {vsn, VsnList} = lists:keyfind(vsn, 1, Attrs), case VsnList of [Vsn] when Vsn =:= Version -> {true, M}; Vsn when Vsn =:= Version -> {true, M}; _Else -> {false, M} end catch error:undef -> ?log(debug, "event=module_not_found module=~s", [M]), {false, M}; error:badarg -> ?log(debug, "event=invalid_module_definition module=\"~p\"", [M]), {false, M} end; check_module_version_compat(M) -> {true, M}.