-module(opcua_server_session_manager). -behaviour(gen_server). %%% INCLUDES %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% -include_lib("kernel/include/logger.hrl"). -include_lib("stdlib/include/assert.hrl"). -include("opcua.hrl"). -include("opcua_internal.hrl"). %%% EXPORTS %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% API Functions -export([start_link/1]). -export([handle_request/2]). %% Behaviour gen_server callback functions -export([init/1]). -export([handle_call/3]). -export([handle_cast/2]). -export([handle_info/2]). -export([terminate/2]). -export([code_change/3]). %%% MACRO %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% -define(SERVER, ?MODULE). %%% TYPES %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% -record(session, { pid :: pid(), auth :: opcua:node_id(), ref :: reference() }). -record(state, { next_session_id = 1 :: pos_integer(), sessions = #{} :: #{pid() => #session{}}, auth_lookup = #{} :: #{binary() => pid()} }). %%% API FUNCTIONS %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% start_link(Opts) -> gen_server:start_link({local, ?SERVER}, ?MODULE, Opts, []). handle_request(Conn, #uacp_message{node_id = #opcua_node_id{value = 459}} = Req) -> Conn2 = opcua_keychain:shareable(Conn), case gen_server:call(?SERVER, {create_session, Conn2, Req}) of {error, _Reason} = Error -> Error; {created, Resp, Conn3, SessPid} -> Conn4 = opcua_connection:merge(Conn, Conn3), {created, Resp, Conn4, SessPid} end; handle_request(Conn, #uacp_message{} = Req) -> Conn2 = opcua_keychain:shareable(Conn), case gen_server:call(?SERVER, {forward_request, Conn2, Req}) of {error, _Reason} = Error -> Error; {Tag, Resp, #uacp_connection{} = Conn3, SessPid} -> Conn4 = opcua_connection:merge(Conn, Conn3), {Tag, Resp, Conn4, SessPid} end. %%% BEHAVIOUR gen_server CALLBACK FUNCTIONS %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% init(Opts) -> ?LOG_DEBUG("OPCUA session manager process starting with options: ~p", [Opts]), {ok, #state{}}. handle_call({create_session, Conn, Req}, _From, State) -> case create_session(State, Conn, Req) of {error, _Reason} = Error -> {reply, Error, State}; {Result, State2} -> {reply, Result, State2} end; handle_call({forward_request, Conn, Req}, _From, State) -> case forward_request(State, Conn, Req) of {error, _Reason} = Error -> {reply, Error, State}; {Result, State2} -> {reply, Result, State2} end; handle_call(Req, From, State) -> ?LOG_WARNING("Unexpected gen_server call from ~p: ~p", [From, Req]), {reply, {error, unexpected_call}, State}. handle_cast(Req, State) -> ?LOG_WARNING("Unexpected gen_server cast: ~p", [Req]), {noreply, State}. handle_info({'DOWN', MonRef, process, SessPid, _Info}, State) -> {noreply, session_del(State, SessPid, MonRef)}; handle_info(Msg, State) -> ?LOG_WARNING("Unexpected gen_server message: ~p", [Msg]), {noreply, State}. terminate(Reason, _State) -> ?LOG_DEBUG("OPCUA session manager process terminating: ~p", [Reason]), ok. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%% INTERNAL FUNCTIONS %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% generate_session_auth_token() -> #opcua_node_id{type = opaque, value = crypto:strong_rand_bytes(32)}. next_session_node_id(#state{next_session_id = Id} = State) -> {#opcua_node_id{ns = 232, value = Id}, State#state{next_session_id = Id + 1}}. create_session(State, Conn, Req) -> %TODO: Probably check the request header... {SessNodeId, State2} = next_session_node_id(State), AuthToken = generate_session_auth_token(), case opcua_server_session_pool_sup:start_session(SessNodeId, AuthToken) of {error, _Reason} = Error -> Error; {ok, SessPid} -> case opcua_server_session:handle_request(Conn, Req, SessPid) of {error, _Reason} = Error -> Error; {created, Resp, Conn2} -> State3 = session_add(State2, SessPid, AuthToken), {{created, Resp, Conn2, SessPid}, State3} end end. forward_request(State, Conn, #uacp_message{payload = Msg} = Req) -> #{request_header := #{authentication_token := AuthToken}} = Msg, case session_find_by_auth(State, AuthToken) of error -> {error, bad_session_id_invalid}; {ok, #session{pid = SessPid}} -> case opcua_server_session:handle_request(Conn, Req, SessPid) of {error, _Reason} = Error -> Error; {Tag, Resp, Conn2} when is_atom(Tag) -> {{Tag, Resp, Conn2, SessPid}, State} end end. session_find_by_auth(State, Auth) -> #state{sessions = Sessions, auth_lookup = AuthLookup} = State, case maps:find(Auth, AuthLookup) of error -> error; {ok, SessPid} -> maps:find(SessPid, Sessions) end. session_add(State, Pid, Auth) -> #state{sessions = Sessions, auth_lookup = AuthLookup} = State, ?assert(not maps:is_key(Pid, Sessions)), ?assert(not maps:is_key(Auth, AuthLookup)), MonRef = erlang:monitor(process, Pid), SessRec = #session{pid = Pid, auth = Auth, ref = MonRef}, State#state{ sessions = Sessions#{Pid => SessRec}, auth_lookup = AuthLookup#{Auth => Pid} }. session_del(State, Pid, MonRef) -> #state{sessions = Sessions, auth_lookup = AuthLookup} = State, case maps:take(Pid, Sessions) of error -> State; {#session{pid = Pid, ref = MonRef, auth = Auth}, Sessions2} -> State#state{ sessions = Sessions2, auth_lookup = maps:remove(Auth, AuthLookup) }; {#session{pid = Pid, ref = OtherMonRef, auth = Auth}, Sessions2} -> erlang:demonitor(OtherMonRef, [flush]), State#state{ sessions = Sessions2, auth_lookup = maps:remove(Auth, AuthLookup) } end.