diff --git a/include/iot_tables.hrl b/include/iot_tables.hrl new file mode 100644 index 0000000..8d118a9 --- /dev/null +++ b/include/iot_tables.hrl @@ -0,0 +1,16 @@ +%%%------------------------------------------------------------------- +%%% @author anlicheng +%%% @copyright (C) 2026, +%%% @doc +%%% +%%% @end +%%% Created : 24. 4月 2026 16:54 +%%%------------------------------------------------------------------- +-author("anlicheng"). + +-record(efka_client, { + uuid :: binary(), + token_hash :: binary(), + salt :: binary(), + timestamp = 0 :: integer() +}). \ No newline at end of file diff --git a/src/iot_app.erl b/src/iot_app.erl index d89b627..6f6aae4 100644 --- a/src/iot_app.erl +++ b/src/iot_app.erl @@ -34,6 +34,7 @@ start_mnesia() -> ok = ensure_mnesia_schema(), %% 启动数据库 ok = mnesia:start(), + ok = ensure_mnesia_tables(), Tables = mnesia:system_info(tables), logger:debug("[iot_app] tables: ~p", [Tables]), %% 创建数据库表 @@ -110,3 +111,19 @@ ensure_mnesia_schema() -> throw({init_schema, Error}) end end. + +-spec ensure_mnesia_tables() -> ok. +ensure_mnesia_tables() -> + ok = ensure_efka_client_table(), + ok = mnesia:wait_for_tables([efka_client], 5000), + ok. + +-spec ensure_efka_client_table() -> ok. +ensure_efka_client_table() -> + case efka_client_store:ensure_table() of + ok -> + ok; + {error, Reason} -> + logger:debug("[iot_app] create efka_client table failed with error: ~p", [Reason]), + throw({init_tables, {efka_client, Reason}}) + end. diff --git a/src/mnesia/efka_client_store.erl b/src/mnesia/efka_client_store.erl new file mode 100644 index 0000000..508b331 --- /dev/null +++ b/src/mnesia/efka_client_store.erl @@ -0,0 +1,121 @@ +%%%------------------------------------------------------------------- +%%% @doc efka_client table store based on mnesia. +%%%------------------------------------------------------------------- +-module(efka_client_store). + +-include("iot_tables.hrl"). + +-export([ensure_table/0, register/2, get/1, update/1, delete/1, list/0]). +-export([auth/2]). + +-define(TOKEN_HASH_ITERATIONS, 32). +-define(TOKEN_HASH_BYTES, 32). + +-spec ensure_table() -> ok | {error, term()}. +ensure_table() -> + case mnesia:create_table(efka_client, [ + {attributes, record_info(fields, efka_client)}, + {type, set}, + {record_name, efka_client}, + {disc_copies, [node()]} + ]) of + {atomic, ok} -> + ok; + {aborted, {already_exists, efka_client}} -> + ok; + {aborted, Reason} -> + {error, Reason} + end. + +-spec auth(UUID :: binary(), Token :: binary()) -> boolean(). +auth(UUID, Token) when is_binary(UUID), UUID =/= <<>>, is_binary(Token) -> + case mnesia:dirty_read(efka_client, UUID) of + [] -> + false; + [#efka_client{token_hash = TokenHash, salt = Salt}] -> + hash_token(Token, Salt) =:= TokenHash + end. + +-spec register(UUID :: binary(), Token :: binary()) -> ok | {error, term()}. +register(UUID, Token) when is_binary(UUID), UUID =/= <<>>, is_binary(Token) -> + case mnesia:transaction(fun() -> + case mnesia:read(efka_client, UUID, write) of + [] -> + %% 1. 生成 16 字节随机盐 + Salt = crypto:strong_rand_bytes(16), + TokenHash = hash_token(Token, Salt), + ok = mnesia:write(#efka_client{uuid = UUID, token_hash = TokenHash, salt = Salt, timestamp = iot_util:timestamp()}), + ok; + [_] -> + {error, already_exists} + end + end) of + {atomic, ok} -> + ok; + {atomic, {error, Reason}} -> + {error, Reason}; + {aborted, Reason} -> + {error, Reason} + end. + +-spec get(binary()) -> {ok, #efka_client{}} | not_found | {error, term()}. +get(UUID) when is_binary(UUID), UUID =/= <<>> -> + case mnesia:transaction(fun() -> + case mnesia:read(efka_client, UUID, read) of + [Client = #efka_client{}] -> + {ok, Client}; + [] -> + not_found + end + end) of + {atomic, Result} -> + Result; + {aborted, Reason} -> + {error, Reason} + end. + +-spec update(#efka_client{}) -> ok | {error, term()}. +update(Client = #efka_client{uuid = UUID}) when is_binary(UUID), UUID =/= <<>> -> + case mnesia:transaction(fun() -> + case mnesia:read(efka_client, UUID, write) of + [_] -> + ok = mnesia:write(Client), + ok; + [] -> + {error, not_found} + end + end) of + {atomic, ok} -> + ok; + {atomic, {error, Reason}} -> + {error, Reason}; + {aborted, Reason} -> + {error, Reason} + end. + +-spec delete(binary()) -> ok | {error, term()}. +delete(UUID) when is_binary(UUID), UUID =/= <<>> -> + case mnesia:transaction(fun() -> + ok = mnesia:delete({efka_client, UUID}), + ok + end) of + {atomic, ok} -> + ok; + {aborted, Reason} -> + {error, Reason} + end. + +-spec list() -> {ok, [#efka_client{}]} | {error, term()}. +list() -> + case mnesia:transaction(fun() -> + mnesia:foldl(fun(Client = #efka_client{}, Acc) -> [Client | Acc] end, [], efka_client) + end) of + {atomic, Clients} -> + {ok, lists:reverse(Clients)}; + {aborted, Reason} -> + {error, Reason} + end. + +-spec hash_token(binary(), binary()) -> binary(). +hash_token(Token, Salt) when is_binary(Token), is_binary(Salt) -> + crypto:pbkdf2_hmac(sha256, Token, Salt, ?TOKEN_HASH_ITERATIONS, ?TOKEN_HASH_BYTES). \ No newline at end of file diff --git a/src/transport/tcp/ssl_channel.erl b/src/transport/tcp/ssl_channel.erl index fbd833c..c976146 100644 --- a/src/transport/tcp/ssl_channel.erl +++ b/src/transport/tcp/ssl_channel.erl @@ -150,7 +150,7 @@ handle_info({stop, Reason}, State) -> %% 主机进程挂掉的时候，需要关闭掉链接 handle_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, host_pid = HostPid}) -> logger:debug("[ws_channel] uuid: ~p, channel will close because host exited with reason: ~p", [UUID, Reason]), - {stop, State}; + {stop, Reason, State}; handle_info({ssl, Socket, <>}, State = #state{transport = Transport, socket = Socket}) -> @@ -228,7 +228,7 @@ handle_request_frame(#'RequestFrame'{packet_id = PacketId, case iot_api_client:get_host_by_uuid(UUID) of undefined -> logger:warning("[ws_channel] uuid: ~p, user: ~p, host not found", [UUID, Username]), - {stop, State}; + {stop, normal, State}; {ok, _} -> %% 尝试启动主机的服务进程 {ok, HostPid} = iot_host_sup:ensured_host_started(UUID), @@ -245,20 +245,20 @@ handle_request_frame(#'RequestFrame'{packet_id = PacketId, {error, Reason} when is_binary(Reason) -> send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyFrame.Error'{code = 2, message = Reason}}), logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]), - {stop, State} + {stop, Reason, State} end end; {error, Reason} -> send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyFrame.Error'{code = 2, message = Reason}}), logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed, reason: ~p", [UUID, Username, Reason]), - {stop, State} + {stop, Reason, State} end; handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = {container_request, ContainerRequest}}, _Transport, _Socket, State) -> logger:warning("[ws_channel] unsupported request message type: container_request, packet_id: ~p, request: ~p", [PacketId, ContainerRequest]), - {stop, State}; + {stop, normal, State}; handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = undefined}, _Transport, _Socket, State) -> logger:warning("[ws_channel] empty request frame, packet_id: ~p", [PacketId]), - {stop, State}. + {stop, normal, State}. -spec handle_cast_frame(message_pb:'CastFrame'(), undefined | pid(), #state{}) -> {noreply, #state{}}.