fix host
This commit is contained in:
anlicheng
parent
e5dc938c13
commit
522c60f355
@ -80,11 +80,12 @@ handle_request("POST", "/host/publish_command", _,
|
||||
},
|
||||
BinReply = jiffy:encode(append_service_name(PostParams, Reply), [force_utf8]),
|
||||
|
||||
case iot_host:build_command(Pid, CommandType, BinReply) of
|
||||
case iot_host:aes_encode(Pid, CommandType, BinReply) of
|
||||
{error, Reason} when is_binary(Reason) ->
|
||||
task_logs_bo:change_status(TaskId, ?TASK_STATUS_FAILED),
|
||||
{ok, 200, iot_util:json_error(400, Reason)};
|
||||
{ok, Topic, BinCommand} ->
|
||||
{ok, BinCommand} ->
|
||||
Topic = iot_host:downstream_topic(UUID),
|
||||
case iot_mqtt_publisher:publish(Topic, BinCommand, 2) of
|
||||
{ok, Ref} ->
|
||||
receive
|
||||
@ -123,10 +124,11 @@ handle_request("POST", "/host/activate", _, #{<<"uuid">> := UUID, <<"auth">> :=
|
||||
BinReply = jiffy:encode(Reply, [force_utf8]),
|
||||
|
||||
CommandType = 8,
|
||||
case iot_host:build_command(Pid, CommandType, BinReply) of
|
||||
case iot_host:rsa_encode(Pid, CommandType, BinReply) of
|
||||
{error, Reason} when is_binary(Reason) ->
|
||||
{ok, 200, iot_util:json_error(401, Reason)};
|
||||
{ok, Topic, BinCommand} ->
|
||||
{ok, BinCommand} ->
|
||||
Topic = iot_host:downstream_topic(UUID),
|
||||
case iot_mqtt_publisher:publish(Topic, BinCommand, 2) of
|
||||
{ok, Ref} ->
|
||||
receive
|
||||
|
||||
@ -18,7 +18,7 @@
|
||||
|
||||
%% API
|
||||
-export([start_link/2, get_name/1, get_pid/1, handle/2, reload/1, activate/1]).
|
||||
-export([get_metric/1, build_command/3, downstream_topic/1, upstream_topic/1, get_aes/1, make_assoc/1]).
|
||||
-export([get_metric/1, aes_encode/3, downstream_topic/1, upstream_topic/1, get_aes/1, make_assoc/1, rsa_encode/3]).
|
||||
|
||||
%% gen_server callbacks
|
||||
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
|
||||
@ -77,6 +77,11 @@ downstream_topic(UUID) when is_binary(UUID) ->
|
||||
upstream_topic(UUID) when is_binary(UUID) ->
|
||||
<<"host/upstream/", UUID/binary>>.
|
||||
|
||||
-spec rsa_encode(Pid :: pid(), CommandType :: integer(), PlainText :: binary()) ->
|
||||
{ok, EncText :: binary()} | {error, Reason :: binary()}.
|
||||
rsa_encode(Pid, CommandType, PlainText) when is_pid(Pid), is_integer(CommandType), is_binary(PlainText) ->
|
||||
gen_server:call(Pid, {rsa_encode, CommandType, PlainText}).
|
||||
|
||||
%% 处理消息
|
||||
-spec handle(Pid :: pid(), Payload :: binary() | map()) -> no_return().
|
||||
handle(Pid, Payload) when is_pid(Pid), is_binary(Payload); is_map(Payload) ->
|
||||
@ -87,6 +92,7 @@ handle(Pid, Payload) when is_pid(Pid), is_binary(Payload); is_map(Payload) ->
|
||||
reload(Pid) when is_pid(Pid) ->
|
||||
gen_server:call(Pid, reload).
|
||||
|
||||
-spec get_aes(Pid :: pid()) -> Aes :: binary().
|
||||
get_aes(Pid) when is_pid(Pid) ->
|
||||
gen_server:call(Pid, get_aes).
|
||||
|
||||
@ -103,10 +109,10 @@ activate(Pid) when is_pid(Pid) ->
|
||||
get_metric(Pid) when is_pid(Pid) ->
|
||||
gen_server:call(Pid, get_metric).
|
||||
|
||||
-spec build_command(Pid :: pid(), CommandType :: integer(), Params :: binary()) ->
|
||||
{ok, Topic :: binary(), Command :: binary()} | {error, Reason :: any()}.
|
||||
build_command(Pid, CommandType, Params) when is_pid(Pid), is_integer(CommandType), is_binary(Params) ->
|
||||
gen_server:call(Pid, {build_command, CommandType, Params}).
|
||||
-spec aes_encode(Pid :: pid(), CommandType :: integer(), Params :: binary()) ->
|
||||
{ok, Command :: binary()} | {error, Reason :: any()}.
|
||||
aes_encode(Pid, CommandType, Params) when is_pid(Pid), is_integer(CommandType), is_binary(Params) ->
|
||||
gen_server:call(Pid, {aes_encode, CommandType, Params}).
|
||||
|
||||
%% @doc Spawns the server and registers the local name (unique)
|
||||
-spec(start_link(Name :: atom(), UUID :: binary()) ->
|
||||
@ -152,6 +158,14 @@ handle_call(get_metric, _From, State = #state{metrics = Metrics}) ->
|
||||
handle_call(get_aes, _From, State = #state{aes = Aes}) ->
|
||||
{reply, {ok, Aes}, State};
|
||||
|
||||
%% 实现rsa加密
|
||||
handle_call({rsa_encode, _, _}, _From, State = #state{has_session = false}) ->
|
||||
{reply, {error, <<"会话未建立,发送命令失败"/utf8>>}, State};
|
||||
handle_call({rsa_encode, CommandType, PlainText}, _From, State = #state{pub_key = PubKey}) ->
|
||||
Reply = iot_cipher_rsa:encode(PlainText, PubKey),
|
||||
|
||||
{reply, {ok, <<CommandType:8, Reply/binary>>}, State};
|
||||
|
||||
handle_call({make_assoc, ReceivePid}, _From, State = #state{uuid = UUID, increment_id = IncrementId, assoc_map = AssocMap, aes = Aes}) ->
|
||||
BinIncrementId = erlang:integer_to_binary(IncrementId),
|
||||
Assoc = <<UUID/binary, ":assoc:", BinIncrementId/binary>>,
|
||||
@ -176,13 +190,12 @@ handle_call(activate, _From, State = #state{uuid = UUID, is_activated = IsActiva
|
||||
{reply, ok, State#state{is_activated = true}};
|
||||
|
||||
%% 创建命令
|
||||
handle_call({build_command, _, _}, _From, State = #state{has_session = false}) ->
|
||||
handle_call({aes_encode, _, _}, _From, State = #state{has_session = false}) ->
|
||||
{reply, {error, <<"会话未建立,发送命令失败"/utf8>>}, State};
|
||||
handle_call({build_command, CommandType, Command}, _From, State = #state{aes = AES, uuid = UUID, has_session = true}) ->
|
||||
DownstreamTopic = downstream_topic(UUID),
|
||||
handle_call({aes_encode, CommandType, Command}, _From, State = #state{aes = AES, uuid = UUID, has_session = true}) ->
|
||||
EncCommand = iot_cipher_aes:encrypt(AES, Command),
|
||||
|
||||
{reply, {ok, DownstreamTopic, <<CommandType:8, EncCommand/binary>>}, State};
|
||||
{reply, {ok, <<CommandType:8, EncCommand/binary>>}, State};
|
||||
|
||||
handle_call(Info, _From, State = #state{}) ->
|
||||
lager:debug("[iot_host] handle info: ~p", [Info]),
|
||||
|
||||
@ -200,6 +200,46 @@ handle_info({publish, #{payload := Payload, qos := Qos, topic := FromTopic}},
|
||||
lager:debug("[host_mocker] auth failed")
|
||||
end;
|
||||
|
||||
%% 8 是通过rsa加密的数据，里面保存了aes；后续的所有通讯都是通过aes加密的
|
||||
<<8:8, Command0/binary>> ->
|
||||
Command = iot_cipher_rsa:decode(Command0, PrivateKey),
|
||||
case jiffy:decode(Command, [return_maps]) of
|
||||
#{<<"aes">> := Aes, <<"auth">> := true, <<"reply">> := #{<<"topic">> := Topic, <<"assoc">> := Assoc}} ->
|
||||
Msg = jiffy:encode(#{
|
||||
<<"code">> => 1,
|
||||
<<"message">> => "",
|
||||
<<"assoc">> => Assoc
|
||||
}, [force_utf8]),
|
||||
|
||||
{ok, Ref} = iot_mqtt_publisher:publish(Topic, Msg, 1),
|
||||
receive
|
||||
{ok, Ref, PacketId} ->
|
||||
lager:debug("[host_mocker] send reply success, packet_id: ~p", [PacketId]);
|
||||
{error, Reason} ->
|
||||
lager:debug("[host_mocker] send reply failed, reason: ~p", [Reason])
|
||||
end,
|
||||
|
||||
{noreply, State#state{aes = Aes}};
|
||||
|
||||
#{<<"aes">> := Aes, <<"auth">> := false, <<"reply">> := #{<<"topic">> := Topic, <<"assoc">> := Assoc}} ->
|
||||
Msg = jiffy:encode(#{
|
||||
<<"code">> => 1,
|
||||
<<"message">> => "",
|
||||
<<"assoc">> => Assoc
|
||||
}, [force_utf8]),
|
||||
|
||||
{ok, Ref} = iot_mqtt_publisher:publish(Topic, Msg, 1),
|
||||
receive
|
||||
{ok, Ref, PacketId} ->
|
||||
lager:debug("[host_mocker] send reply success, packet_id: ~p", [PacketId]);
|
||||
{error, Reason} ->
|
||||
lager:debug("[host_mocker] send reply failed, reason: ~p", [Reason])
|
||||
end,
|
||||
|
||||
{noreply, State#state{aes = <<>>}}
|
||||
end;
|
||||
|
||||
%% 处理其他指令
|
||||
<<Type:8, Command0/binary>> ->
|
||||
Command = iot_cipher_aes:decrypt(Aes0, Command0),
|
||||
CommandJson = jiffy:decode(Command, [return_maps]),
|
||||
@ -378,36 +418,7 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
|
||||
%%% Internal functions
|
||||
%%%===================================================================
|
||||
|
||||
handle_command(8, #{<<"aes">> := Aes, <<"auth">> := true, <<"reply">> := #{<<"topic">> := Topic, <<"assoc">> := Assoc}}, State) ->
|
||||
Msg = jiffy:encode(#{
|
||||
<<"code">> => 1,
|
||||
<<"message">> => "",
|
||||
<<"assoc">> => Assoc
|
||||
}, [force_utf8]),
|
||||
|
||||
{ok, Ref} = iot_mqtt_publisher:publish(Topic, Msg, 1),
|
||||
receive
|
||||
{ok, Ref, PacketId} ->
|
||||
lager:debug("[host_mocker] send reply success, packet_id: ~p", [PacketId]);
|
||||
{error, Reason} ->
|
||||
lager:debug("[host_mocker] send reply failed, reason: ~p", [Reason])
|
||||
end,
|
||||
|
||||
State#state{aes = Aes};
|
||||
handle_command(8, #{<<"aes">> := Aes, <<"auth">> := false, <<"reply">> := #{<<"topic">> := Topic, <<"assoc">> := Assoc}}, State) ->
|
||||
Msg = jiffy:encode(#{
|
||||
<<"code">> => 1,
|
||||
<<"message">> => "",
|
||||
<<"assoc">> => Assoc
|
||||
}, [force_utf8]),
|
||||
|
||||
{ok, Ref} = iot_mqtt_publisher:publish(Topic, Msg, 1),
|
||||
receive
|
||||
{ok, Ref, PacketId} ->
|
||||
lager:debug("[host_mocker] send reply success, packet_id: ~p", [PacketId]);
|
||||
{error, Reason} ->
|
||||
lager:debug("[host_mocker] send reply failed, reason: ~p", [Reason])
|
||||
end,
|
||||
|
||||
State#state{aes = Aes}.
|
||||
handle_command(8, Info, State) ->
|
||||
lager:debug("[host_mocker] command is: ~p", [Info]),
|
||||
State.
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user