fix heartbeat

docs/heartbeat.md

@@ -1,11 +1,44 @@
 ## 心跳机制
 * 边缘主机通过心跳机制来判断主机是否存活(解决弱网环境下websocket链接会经常断开的问题)
-* 边缘主机每隔5秒发送一次心跳包，服务端每隔2分钟检测一下，判断是否有收到心跳包；如果没有收到则认为主机离线
+* 边缘主机定期发送 UDP 心跳包，服务端按照 `iot_host` 的心跳检测周期判断是否有收到心跳包。
+* 如果 UDP 心跳丢失但 SSL channel 仍然存在，服务端不会把 host 标记为离线。
+* 如果 UDP 心跳丢失且 SSL channel 也不存在，服务端会把 host 标记为离线。
 
 ### udp服务器
 * 端口: 18080
 
 ### 心跳包格式
-*  <<Len:2, HostUUID/binary>>
+旧格式 `<<Len:16, HostUUID/binary>>` 已不再接受。
-* Len表示HostUUID对应的字节数，Len本身占用2字节长度(HostUUID不一定是固定长度，因此需要标注)
+
-* 注解：采用这种格式是为了方便后续扩展别的心跳信息字段
+当前心跳包格式：
+
+```erlang
+<<
+    Version:8,
+    UuidLen:16,
+    UUID:UuidLen/binary,
+    Timestamp:64/unsigned-big,
+    Nonce:16/binary,
+    Mac:32/binary
+>>
+```
+
+字段说明：
+
+| 字段 | 说明 |
+| --- | --- |
+| `Version` | 当前固定为 `1`。 |
+| `UuidLen` | `UUID` 的字节数。 |
+| `UUID` | efka 对应的 host UUID。 |
+| `Timestamp` | 秒级 Unix 时间戳。服务端只接受 120 秒时间窗内的心跳。 |
+| `Nonce` | 16 字节随机数，必须加入 HMAC payload，使同一秒内多次心跳的 Mac 不同。当前服务端不保存 nonce。 |
+| `Mac` | HMAC-SHA256 结果，32 字节。 |
+
+`Mac` 的计算内容为前面所有字段：
+
+```erlang
+Payload = <<Version:8, UuidLen:16, UUID:UuidLen/binary, Timestamp:64/unsigned-big, Nonce:16/binary>>,
+Mac = crypto:mac(hmac, sha256, HeartbeatSecret, Payload)
+```
+
+服务端在 `efka_client_store:verify_heartbeat/4` 中校验时间窗和 HMAC。当前服务端使用 `efka_client_store` 中保存的 `token_hash` 作为 `HeartbeatSecret`，避免在 iot 侧保存明文 auth token。

src/mnesia/efka_client_store.erl

@@ -6,10 +6,11 @@
 -include("iot_tables.hrl").
 
 -export([ensure_table/0, register/2, get/1, update/1, delete/1, list/0]).
--export([auth/2]).
+-export([auth/2, verify_heartbeat/4]).
 
 -define(TOKEN_HASH_ITERATIONS, 32).
 -define(TOKEN_HASH_BYTES, 32).
+-define(HEARTBEAT_TIMESTAMP_WINDOW, 120).
 
 -spec ensure_table() -> ok | {error, term()}.
 ensure_table() ->
@@ -36,6 +37,22 @@ auth(UUID, Token) when is_binary(UUID), UUID =/= <<>>, is_binary(Token) ->
             hash_token(Token, Salt) =:= TokenHash
     end.
 
+-spec verify_heartbeat(UUID :: binary(), Timestamp :: integer(), Payload :: binary(), Mac :: binary()) -> boolean().
+verify_heartbeat(UUID, Timestamp, Payload, Mac)
+  when is_binary(UUID), UUID =/= <<>>, is_integer(Timestamp), is_binary(Payload), is_binary(Mac) ->
+    case valid_heartbeat_timestamp(Timestamp) of
+        true ->
+            case mnesia:dirty_read(efka_client, UUID) of
+                [] ->
+                    false;
+                [#efka_client{token_hash = TokenHash}] ->
+                    ExpectedMac = crypto:mac(hmac, sha256, TokenHash, Payload),
+                    ExpectedMac =:= Mac
+            end;
+        false ->
+            false
+    end.
+
 -spec register(UUID :: binary(), Token :: binary()) -> ok | {error, term()}.
 register(UUID, Token) when is_binary(UUID), UUID =/= <<>>, is_binary(Token), byte_size(Token) >= 32 ->
     case mnesia:transaction(fun() ->
@@ -119,3 +136,8 @@ list() ->
 -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).
+
+-spec valid_heartbeat_timestamp(integer()) -> boolean().
+valid_heartbeat_timestamp(Timestamp) ->
+    Now = iot_util:current_time(),
+    Timestamp =< Now andalso Now - Timestamp =< ?HEARTBEAT_TIMESTAMP_WINDOW.

src/transport/udp/udp_server.erl

@@ -12,16 +12,47 @@
 %% API
 -export([start_link/2, loop/2]).
 
+-define(HEARTBEAT_VERSION, 1).
+-define(HEARTBEAT_NONCE_BYTES, 16).
+-define(HEARTBEAT_MAC_BYTES, 32).
+
 start_link(Transport, Peer) ->
     {ok, spawn_link(?MODULE, loop, [Transport, Peer])}.
 
 loop(Transport = {udp, Server, _Sock}, Peer) ->
     receive
-        {datagram, Server, <<Len:16, HostUUID:Len/binary>>} ->
-            Pid = iot_host:get_pid(HostUUID),
-            iot_host:heartbeat(Pid),
-            loop(Transport, Peer);
         {datagram, Server, Packet} ->
-            logger:warning("[udp_server] ignore invalid heartbeat packet from peer: ~p, packet: ~p", [Peer, Packet]),
+            handle_heartbeat_packet(Packet, Peer),
+            loop(Transport, Peer);
+        Other ->
+            logger:warning("[udp_server] ignore unknown message from peer: ~p, message: ~p", [Peer, Other]),
             loop(Transport, Peer)
     end.
+
+-spec handle_heartbeat_packet(binary(), term()) -> ok.
+handle_heartbeat_packet(Packet, Peer) when is_binary(Packet) ->
+    case decode_heartbeat_packet(Packet) of
+        {ok, UUID, Timestamp, Payload, Mac} ->
+            case efka_client_store:verify_heartbeat(UUID, Timestamp, Payload, Mac) of
+                true ->
+                    Pid = iot_host:get_pid(UUID),
+                    iot_host:heartbeat(Pid);
+                false ->
+                    logger:warning("[udp_server] ignore invalid heartbeat auth from peer: ~p, uuid: ~p", [Peer, UUID]),
+                    ok
+            end;
+        error ->
+            logger:warning("[udp_server] ignore invalid heartbeat packet from peer: ~p, packet: ~p", [Peer, Packet]),
+            ok
+    end.
+
+-spec decode_heartbeat_packet(binary()) ->
+    {ok, UUID :: binary(), Timestamp :: integer(), Payload :: binary(), Mac :: binary()} | error.
+decode_heartbeat_packet(Packet = <<?HEARTBEAT_VERSION:8, UUIDLen:16, UUID:UUIDLen/binary,
+    Timestamp:64/unsigned-big, _Nonce:?HEARTBEAT_NONCE_BYTES/binary, _Mac:?HEARTBEAT_MAC_BYTES/binary>>)
+  when UUIDLen > 0 ->
+    PayloadSize = byte_size(Packet) - ?HEARTBEAT_MAC_BYTES,
+    <<Payload:PayloadSize/binary, Mac:?HEARTBEAT_MAC_BYTES/binary>> = Packet,
+    {ok, UUID, Timestamp, Payload, Mac};
+decode_heartbeat_packet(_Packet) ->
+    error.