anlicheng/iot_cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix ssl_channel

Browse Source
This commit is contained in:
anlicheng 2026-05-09 14:42:29 +08:00
parent 509d679d70
commit 67b14096c7
2 changed files with 68 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/efka_iot_protocol.md
View File

@ -37,7 +37,7 @@
## 鉴权请求
初始连接由 `efka` 发起鉴权 request：
初始连接由 `efka` 发起鉴权 request。每条 TLS 连接只允许一次鉴权；`iot` 侧鉴权成功后会在 `ssl_channel` 标记该连接已鉴权，如果同一连接再次发送 `auth_request`，`iot` 会直接关闭连接。
```erlang
{request, Ref, {auth_request, #{

89
src/transport/tcp/ssl_channel.erl
View File

@ -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])).