fix efka

docs/efka_iot_protocol.md

@@ -33,7 +33,6 @@
 
 `command` 和 `command_response` 的 `Domain` 表示业务域，目前支持：
 
-- `auth`
 - `container`
 
 ## 鉴权请求
@@ -52,37 +51,19 @@
 
 ```erlang
 {response, Ref, {auth_response, ok}}
-{response, Ref, {auth_response, {error, {denied, Reason}}}}
 {response, Ref, {auth_response, {error, {failed, Reason}}}}
 ```
 
 处理语义：
 
 - `ok`：`efka` 进入 `activated` 状态。
-- `{error, {denied, Reason}}`：`efka` 进入 `restricted` 状态，不能正常上报数据，但仍可接收部分命令。
 - `{error, {failed, Reason}}`：鉴权失败，连接关闭后重连。
 
-## 授权控制命令
+## 授权控制
 
-`iot` 对 `efka` 的授权控制使用 command 语义：
+`/host/activate` 只修改 `iot` 本地和持久化的 host 授权状态，不再向 `efka` 下发 auth command。`efka` 可以继续保持连接并发送数据，是否处理这些数据由 `iot_host` 当前状态决定。
 
-```erlang
+因此当前协议没有 `{command, Ref, {auth, ...}}` 和 `{command_response, Ref, {auth, ...}}`。授权关闭时，`iot_host` 保持 channel 在线，但不处理上报数据；授权重新打开后，已在线的 channel 可以继续使用。
-{command, Ref, {auth, activate}}
-{command, Ref, {auth, deactivate}}
-```
-
-`efka` 回复：
-
-```erlang
-{command_response, Ref, {auth, ok}}
-{command_response, Ref, {auth, {error, Reason}}}
-```
-
-处理语义：
-
-- `activate`：如果 `efka` 已经是 `activated`，直接回复 `ok`；否则重新发送 `auth_request`，等待鉴权结果后再回复该 command。
-- `deactivate`：`efka` 进入 `restricted` 状态，并回复 `ok`。
-- `iot` 侧会异步提交 auth command 并等待对应 `Ref` 的 `command_response`，再向 `/host/activate` HTTP 调用方返回结果；等待超时为 10 秒，超时返回 `timeout`，无效响应返回 `invalid response`。
 
 ## 容器管理命令
 
@@ -252,5 +233,6 @@ ok
 - 旧容器管理：`{request, Ref, {container_request, ...}}`
 - 旧容器回复：`{response, Ref, {container_response, ...}}`
 - 旧授权控制：`{message, {auth_control, Command}}`
+- 已移除的 auth command：`{command, Ref, {auth, activate | deactivate}}`
 
 如果需要滚动升级，应先增加临时兼容分支或引入协议版本协商。

src/transport/efka_client.erl

@@ -24,9 +24,8 @@
 
 %% 标记当前agent的状态，只有在 activated 状态下才可以正常的发送数据
 -define(STATE_DISCONNECTED, disconnected).
+%% 等待校验中
 -define(STATE_AUTH, auth).
-%% 不能推送消息到服务，但是可以接受服务器的部分指令
--define(STATE_RESTRICTED, restricted).
 %% 激活状态下
 -define(STATE_ACTIVATED, activated).
 
@@ -34,8 +33,6 @@
     socket :: undefined | ssl:sslsocket(),
     %% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系
     auth_ref = undefined :: undefined | reference(),
-    %% iot auth 命令触发重新鉴权时，用它保存 command 的 ref。
-    auth_control_ref = undefined :: undefined | reference(),
     dropped_message_count = 0 :: non_neg_integer()
 }).
 
@@ -127,7 +124,7 @@ handle_event({call, From}, dropped_message_count, _StateName, State = #state{dro
     {keep_state, State, [{reply, From, DroppedCount}]};
 
 %% 异步建立到服务器的连接
-handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State = #state{}) ->
+handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State) ->
     case connect_socket() of
         {ok, Socket} ->
             Ref = make_ref(),
@@ -139,12 +136,11 @@ handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State =
             {keep_state, State#state{socket = undefined}}
     end;
 
-handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = Socket, auth_control_ref = AuthControlRef}) ->
+handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = Socket}) ->
     logger:debug("[efka_client] auth request timeout"),
-    maybe_send_auth_response(Socket, AuthControlRef, {error, timeout}),
     disconnect(Socket),
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, auth_control_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
 
 %% 将缓存中的数据推送到服务器端
 handle_event(info, flush_cache, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
@@ -167,73 +163,56 @@ handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{socket = Socke
     logger:debug("[efka_client] ssl error: ~p", [Reason]),
     disconnect(Socket),
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, auth_control_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
 handle_event(info, {ssl_closed, Socket}, _, State = #state{socket = Socket}) ->
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, auth_control_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
 
 %%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配
 
 %% 容器管理命令由 iot 发起，使用 command/command_response 语义。
-handle_event(internal, {command, Ref, {container, #{action := list}}},
+handle_event(internal, {command, Ref, {container, #{action := list}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_commands:get_containers(),
     send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {command, Ref, {container, #{action := deploy, task_id := TaskId, params := Params}}},
+handle_event(internal, {command, Ref, {container, #{action := deploy, task_id := TaskId, params := Params}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_deploy_manager:deploy(TaskId, Params),
     send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {command, Ref, {container, #{action := start, target := Target}}},
+handle_event(internal, {command, Ref, {container, #{action := start, target := Target}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_commands:start_container(container_target(Target)),
     send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {command, Ref, {container, #{action := stop, target := Target, timeout_seconds := TimeoutSeconds}}},
+handle_event(internal, {command, Ref, {container, #{action := stop, target := Target, timeout_seconds := TimeoutSeconds}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
     send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {command, Ref, {container, #{action := kill, target := Target, signal := Signal}}},
+handle_event(internal, {command, Ref, {container, #{action := kill, target := Target, signal := Signal}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
     send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {command, Ref, {container, #{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes}}},
+handle_event(internal, {command, Ref, {container, #{action := remove, target := Target, force := Force, remove_volumes := RemoveVolumes}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
     send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {command, Ref, {container, #{action := config, target := Target, config := Config}}},
+handle_event(internal, {command, Ref, {container, #{action := config, target := Target, config := Config}}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)),
     send_container_response(Socket, Ref, Reply),
     {keep_state, State};
-handle_event(internal, {command, Ref, {container, Request}}, ?STATE_RESTRICTED, State = #state{socket = Socket}) ->
-    logger:notice("[efka_client] get an invalid command: ~p, agent restricted", [Request]),
-    send_container_response(Socket, Ref, {error, <<"agent restricted">>}),
-    {keep_state, State};
 handle_event(internal, {command, Ref, {container, Request}}, _StateName, State = #state{socket = Socket}) ->
     logger:notice("[efka_client] get an invalid command: ~p, agent invalid", [Request]),
     send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
     {keep_state, State};
 
 %% 处理response
-handle_event(internal, {response, AuthRef, {auth_response, ok}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef, auth_control_ref = AuthControlRef}) ->
+handle_event(internal, {response, AuthRef, {auth_response, ok}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
     logger:debug("[efka_client] auth success"),
-    maybe_send_auth_response(Socket, AuthControlRef, ok),
+    {next_state, ?STATE_ACTIVATED, State#state{auth_ref = undefined}, [{next_event, info, flush_cache}]};
-    {next_state, ?STATE_ACTIVATED, State#state{auth_ref = undefined, auth_control_ref = undefined}, [{next_event, info, flush_cache}]};
+handle_event(internal, {response, AuthRef, {auth_response, {error, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
-handle_event(internal, {response, AuthRef, {auth_response, {error, {denied, Message}}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef, auth_control_ref = AuthControlRef}) ->
-    logger:debug("[efka_client] auth denied, message: ~p", [Message]),
-    maybe_send_auth_response(Socket, AuthControlRef, {error, {denied, Message}}),
-    {next_state, ?STATE_RESTRICTED, State#state{auth_ref = undefined, auth_control_ref = undefined}};
-handle_event(internal, {response, AuthRef, {auth_response, {error, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef, auth_control_ref = AuthControlRef}) ->
     logger:debug("[efka_client] auth failed, reason: ~p", [Reason]),
-    maybe_send_auth_response(Socket, AuthControlRef, {error, Reason}),
     disconnect(Socket),
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, auth_control_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
 handle_event(internal, {response, _Ref, Reply}, StateName, State) ->
     logger:warning("[efka_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
     {keep_state, State};
@@ -241,23 +220,6 @@ handle_event(internal, {command_response, _Ref, Reply}, StateName, State) ->
     logger:warning("[efka_client] ignore unexpected command_response in state ~p: ~p", [StateName, Reply]),
     {keep_state, State};
 
-%% 处理命令
-handle_event(internal, {command, CommandRef, {auth, Cmd}}, StateName, State = #state{socket = Socket}) ->
-    logger:debug("[efka_client] auth cmd: ~p", [Cmd]),
-    case {Cmd, StateName} of
-        {activate, ?STATE_ACTIVATED} ->
-            send_auth_response(Socket, CommandRef, ok),
-            {keep_state, State};
-        {activate, _} ->
-            Ref = make_ref(),
-            AuthPacket = auth_packet(Ref),
-            ok = ssl:send(Socket, AuthPacket),
-            {next_state, ?STATE_AUTH, State#state{auth_ref = Ref, auth_control_ref = CommandRef}, [{state_timeout, 5000, auth_timeout}]};
-        {deactivate, _} ->
-            send_auth_response(Socket, CommandRef, ok),
-            {next_state, ?STATE_RESTRICTED, State#state{auth_ref = undefined, auth_control_ref = undefined}}
-    end;
-
 %% 处理Pub/Sub机制
 handle_event(internal, {message, {pub, #{topic := Topic, qos := Qos, content := Content}}}, ?STATE_ACTIVATED, State) ->
     logger:debug("[efka_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
@@ -327,17 +289,6 @@ send_container_response(Socket, Ref, Reply) ->
     Packet = term_to_binary({command_response, Ref, {container, Reply}}),
     ok = ssl:send(Socket, Packet).
 
--spec send_auth_response(ssl:sslsocket(), reference(), term()) -> ok.
-send_auth_response(Socket, Ref, Reply) ->
-    Packet = term_to_binary({command_response, Ref, {auth, Reply}}),
-    ok = ssl:send(Socket, Packet).
-
--spec maybe_send_auth_response(ssl:sslsocket(), undefined | reference(), term()) -> ok.
-maybe_send_auth_response(_Socket, undefined, _Reply) ->
-    ok;
-maybe_send_auth_response(Socket, Ref, Reply) when is_reference(Ref) ->
-    send_auth_response(Socket, Ref, Reply).
-
 -spec container_target(map()) -> binary().
 container_target(Target) when is_map(Target) ->
     NameBin = to_binary(maps:get(name, Target, <<>>)),