diff --git a/docs/efka_iot_protocol.md b/docs/efka_iot_protocol.md index 8f959ec..3dcef47 100644 --- a/docs/efka_iot_protocol.md +++ b/docs/efka_iot_protocol.md @@ -37,7 +37,7 @@ ## 鉴权请求 -初始连接由 `efka` 发起鉴权 request： +初始连接由 `efka` 发起鉴权 request。每条 TLS 连接只允许一次鉴权；`iot` 侧鉴权成功后会在 `ssl_channel` 标记该连接已鉴权，如果同一连接再次发送 `auth_request`，`iot` 会直接关闭连接。 ```erlang {request, Ref, {auth_request, #{ diff --git a/src/transport/tcp/ssl_channel.erl b/src/transport/tcp/ssl_channel.erl index 3da6bd2..635d0a2 100644 --- a/src/transport/tcp/ssl_channel.erl +++ b/src/transport/tcp/ssl_channel.erl @@ -24,6 +24,7 @@ socket, uuid :: undefined | binary(), + is_authed = false :: boolean(), %% 用户进程id host_pid = undefined, @@ -85,18 +86,27 @@ handle_call(_Request, _From, State) -> %% 发送消息, 基于pub/sub机制 handle_cast({pub, Topic, Qos, Content}, State = #state{transport = Transport, socket = Socket}) -> Packet = term_to_binary({message, {pub, #{topic => Topic, qos => Qos, content => Content}}}), - Transport:send(Socket, Packet), - {noreply, State}; + case Transport:send(Socket, Packet) of + ok -> + {noreply, State}; + {error, Reason} -> + logger:warning("[ssl_channel] send pub failed, reason: ~p", [Reason]), + {stop, {send_failed, Reason}, State} + end; %% iot 请求 efka 时使用 command/command_response 语义。 handle_cast({command_call, ReceiverPid, Ref, Body}, State = #state{transport = Transport, socket = Socket, inflight = Inflight}) -> Packet = term_to_binary({command, Ref, Body}), - Transport:send(Socket, Packet), - - TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {command_timeout, Ref}), - CommandInfo = #inflight_command{receiver_pid = ReceiverPid, timer_ref = TimerRef}, - - {noreply, State#state{inflight = maps:put(Ref, CommandInfo, Inflight)}}. + case Transport:send(Socket, Packet) of + ok -> + TimerRef = erlang:start_timer(?INFLIGHT_TIMEOUT, self(), {command_timeout, Ref}), + CommandInfo = #inflight_command{receiver_pid = ReceiverPid, timer_ref = TimerRef}, + {noreply, State#state{inflight = maps:put(Ref, CommandInfo, Inflight)}}; + {error, Reason} -> + logger:warning("[ssl_channel] send command failed, ref: ~p, reason: ~p", [Ref, Reason]), + deliver_command_error(ReceiverPid, Ref, {send_failed, Reason}), + {stop, {send_failed, Reason}, State} + end. handle_info({timeout, TimerRef, {command_timeout, Ref}}, State = #state{inflight = Inflight}) -> case maps:get(Ref, Inflight, undefined) of @@ -162,6 +172,9 @@ code_change(_OldVsn, State, _Extra) -> %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% -spec handle_request_frame(reference(), tuple(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}. +handle_request_frame(Ref, {auth_request, #{uuid := UUID}}, State = #state{is_authed = true}) -> + logger:warning("[ws_channel] repeated auth request, ref: ~p, uuid: ~p, close channel", [Ref, UUID]), + {stop, repeated_auth, State}; handle_request_frame(Ref, {auth_request, #{uuid := UUID, token := Token, timestamp := Timestamp}}, State = #state{transport = Transport, socket = Socket}) -> logger:debug("[ws_channel] auth uuid: ~p", [UUID]), case auth(Token, UUID, Timestamp) of @@ -169,25 +182,34 @@ handle_request_frame(Ref, {auth_request, #{uuid := UUID, token := Token, timesta case iot_api_client:get_host_by_uuid(UUID) of undefined -> logger:warning("[ws_channel] uuid: ~p, token: ~p, host not found", [UUID, Token]), - {stop, normal, State}; + stop_after_reply(Transport, Socket, Ref, {auth_response, {error, {failed, <<"host not found">>}}}, normal, State); {ok, _} -> %% 尝试启动主机的服务进程 - {ok, HostPid} = iot_host_sup:ensured_host_started(UUID), - case iot_host:attach_channel(HostPid, self()) of - ok -> - erlang:monitor(process, HostPid), - send_reply_frame(Transport, Socket, Ref, {auth_response, ok}), - {noreply, State#state{uuid = UUID, host_pid = HostPid}}; - {error, Reason} when is_binary(Reason) -> - send_reply_frame(Transport, Socket, Ref, {auth_response, {error, {failed, Reason}}}), - logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), - {stop, Reason, State} + case iot_host_sup:ensured_host_started(UUID) of + {ok, HostPid} -> + case iot_host:attach_channel(HostPid, self()) of + ok -> + erlang:monitor(process, HostPid), + case send_reply_frame(Transport, Socket, Ref, {auth_response, ok}) of + ok -> + {noreply, State#state{uuid = UUID, is_authed = true, host_pid = HostPid}}; + {error, Reason} -> + logger:warning("[ws_channel] uuid: ~p, send auth response failed: ~p", [UUID, Reason]), + {stop, {send_failed, Reason}, State} + end; + {error, Reason} when is_binary(Reason) -> + logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), + stop_after_reply(Transport, Socket, Ref, {auth_response, {error, {failed, Reason}}}, Reason, State) + end; + {error, Reason} -> + logger:warning("[ws_channel] uuid: ~p, ensure host failed: ~p", [UUID, Reason]), + ReasonBin = reason_to_binary(Reason), + stop_after_reply(Transport, Socket, Ref, {auth_response, {error, {failed, ReasonBin}}}, Reason, State) end end; {error, Reason} -> - send_reply_frame(Transport, Socket, Ref, {auth_response, {error, {failed, Reason}}}), logger:warning("[ws_channel] uuid: ~p, token: ~p, auth failed, reason: ~p", [UUID, Token, Reason]), - {stop, Reason, State} + stop_after_reply(Transport, Socket, Ref, {auth_response, {error, {failed, Reason}}}, Reason, State) end; handle_request_frame(Ref, {container, ContainerCommand}, State) -> logger:warning("[ws_channel] unsupported request message type: container, ref: ~p, command: ~p", [Ref, ContainerCommand]), @@ -230,11 +252,21 @@ handle_command_response_frame(Ref, Reply, State) -> logger:warning("[ws_channel] unexpected command_response frame, ref: ~p, reply: ~p", [Ref, Reply]), {noreply, State}. --spec send_reply_frame(module(), any(), reference(), tuple()) -> any(). +-spec send_reply_frame(module(), any(), reference(), tuple()) -> ok | {error, term()}. send_reply_frame(Transport, Socket, Ref, Reply) -> Packet = term_to_binary({response, Ref, Reply}), Transport:send(Socket, Packet). +-spec stop_after_reply(module(), any(), reference(), tuple(), term(), #state{}) -> {stop, term(), #state{}}. +stop_after_reply(Transport, Socket, Ref, Reply, StopReason, State) -> + case send_reply_frame(Transport, Socket, Ref, Reply) of + ok -> + {stop, StopReason, State}; + {error, Reason} -> + logger:warning("[ws_channel] send response before stop failed, ref: ~p, reason: ~p", [Ref, Reason]), + {stop, {send_failed, Reason}, State} + end. + -spec decode_command_response({container, ok | {ok, term()} | {error, term()}} | tuple()) -> ok | {ok, term()} | {error, term()}. decode_command_response({container, ok}) -> @@ -257,6 +289,13 @@ deliver_command_response(ReceiverPid, Ref, Reply) when is_pid(ReceiverPid) -> logger:warning("[ws_channel] get command_response: ~p, ref: ~p, but receiver_pid is deaded", [Reply, Ref]) end. +-spec deliver_command_error(undefined | pid(), reference(), term()) -> ok. +deliver_command_error(ReceiverPid, Ref, Reason) when is_pid(ReceiverPid) -> + ReceiverPid ! {command_reply, Ref, {error, Reason}}, + ok; +deliver_command_error(_ReceiverPid, _Ref, _Reason) -> + ok. + -spec reply_command_timeout(undefined | pid(), reference()) -> ok. reply_command_timeout(ReceiverPid, Ref) when is_pid(ReceiverPid) -> ReceiverPid ! {command_reply, Ref, {error, timeout}}, @@ -288,3 +327,9 @@ auth(Token, UUID, Timestamp) when is_binary(Token), is_binary(UUID), is_integer( false -> {error, <<"invalid timestamp">>} end. + +-spec reason_to_binary(term()) -> binary(). +reason_to_binary(Reason) when is_binary(Reason) -> + Reason; +reason_to_binary(Reason) -> + unicode:characters_to_binary(io_lib:format("~p", [Reason])).