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 15:12:34 +08:00
parent 67b14096c7
commit 32cc8367c7
3 changed files with 23 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
docs/efka_iot_protocol.md
View File

@ -47,17 +47,16 @@
}}}
```
`iot` 回复：
`iot` 鉴权成功时回复：
```erlang
{response, Ref, {auth_response, ok}}
{response, Ref, {auth_response, {error, {failed, Reason}}}}
```
处理语义：
- `ok`：`efka` 进入 `activated` 状态。
- `{error, {failed, Reason}}`：鉴权失败，连接关闭后重连。
- 鉴权失败、host 不存在、host 未启动或 attach channel 失败时，`iot` 不返回业务错误响应，直接关闭连接；`efka` 通过 socket close/error 或 auth timeout 进入重连流程。
## 授权控制

10
src/host/iot_host.erl
View File

@ -22,6 +22,7 @@
%% API
-export([start_link/2, get_name/1, get_alias_name/1, get_pid/1, handle/2, activate/2]).
-export([lookup_pid/1]).
-export([get_metric/1, get_status/1, kill/1]).
%% 通讯相关
-export([pub/4, attach_channel/2]).
@ -52,6 +53,15 @@ get_pid(UUID) when is_binary(UUID) ->
Name = get_name(UUID),
gproc:whereis_name({n, l, Name}).
-spec lookup_pid(UUID :: binary()) -> {ok, pid()} | {error, Reason :: any()}.
lookup_pid(UUID) when is_binary(UUID) ->
case get_pid(UUID) of
undefined ->
{error, <<"host not found">>};
HostPid ->
{ok, HostPid}
end.
-spec get_name(UUID :: binary()) -> term().
get_name(UUID) when is_binary(UUID) ->
{iot_host, UUID}.

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

@ -176,40 +176,17 @@ handle_request_frame(Ref, {auth_request, #{uuid := UUID}}, State = #state{is_aut
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
ok ->
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_after_reply(Transport, Socket, Ref, {auth_response, {error, {failed, <<"host not found">>}}}, normal, State);
{ok, _} ->
%% 尝试启动主机的服务进程
case iot_host_sup:ensured_host_started(UUID) of
{ok, HostPid} ->
case iot_host:attach_channel(HostPid, self()) of
ok ->
maybe
ok ?= auth(Token, UUID, Timestamp),
{ok, HostPid} ?= iot_host:lookup_pid(UUID),
ok ?= iot_host:attach_channel(HostPid, self()),
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} ->
logger:warning("[ws_channel] uuid: ~p, token: ~p, auth failed, reason: ~p", [UUID, Token, Reason]),
stop_after_reply(Transport, Socket, Ref, {auth_response, {error, {failed, Reason}}}, Reason, State)
ok ?= send_reply_frame(Transport, Socket, Ref, {auth_response, ok}),
logger:debug("[ws_channel] auth uuid: ~p", [UUID]),
{noreply, State#state{uuid = UUID, is_authed = true, host_pid = HostPid}}
else {error, Reason} ->
logger:warning("[ws_channel] uuid: ~p, auth failed with reason: ~p", [UUID, Reason]),
{stop, 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]),
@ -257,16 +234,6 @@ 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}) ->
@ -327,9 +294,3 @@ 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])).