%%%------------------------------------------------------------------- %%% @author anlicheng %%% @copyright (C) 2024, %%% @doc %%% %%% @end %%% Created : 27. 3月 2024 15:13 %%%------------------------------------------------------------------- -module(sdlan_network). -author("anlicheng"). -include("sdlan.hrl"). -include("sdlan_pb.hrl"). -include_lib("stdlib/include/ms_transform.hrl"). -behaviour(gen_server). %% broadcast, "FF-FF-FF-FF-FF-FF" -define(BROADCAST_MAC, <<16#FF,16#FF,16#FF,16#FF,16#FF,16#FF>>). %% API -export([start_link/2]). -export([get_name/1, get_pid/1, lookup_pid/1, peer_info/3, unregister/3, debug_info/1, get_network_id/1, attach/7, arp_request/2]). -export([forward_by_ets/5, update_hole/7, disable_client/2, get_channel/2, acl_changed/2]). -export([command/4, wait_command_ack/2]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -record(hole, { peer :: {Ip :: inet:ip4_address(), Port :: integer()}, nat_type :: integer() }). %% ip的使用信息, 记录Node的运行时状态信息 -record(endpoint, { channel_pid :: undefined | pid(), channel_ref :: undefined | reference(), transport :: module(), client_id :: binary(), mac :: binary(), ip :: integer(), hostname :: binary(), hole :: undefined | #hole{}, %% 记录ip和ip_v6的映射关系, #{ip_addr :: integer() => {}} v6_info :: undefined | #'SDLV6Info'{}, session_token :: binary(), last_seen :: integer() %% monotonic_time(second), }). -record(state, { network_id :: integer(), name :: binary(), domain :: binary(), ipaddr :: binary(), mask_len :: integer(), owner_id :: integer(), %% 加密算法, 默认为chacha20 algorithm :: binary(), %% 同一个网络下公用的密钥, 采用AES-256加密算法;随机生成 key :: binary(), %% 设置网络带宽 throttle_key :: atom(), endpoint_table :: ets:tid() }). %%%=================================================================== %%% API %%%=================================================================== -spec get_pid(Id :: integer()) -> undefined | pid(). get_pid(Id) when is_integer(Id) -> whereis(get_name(Id)). -spec lookup_pid(Id :: integer()) -> {ok, Pid :: pid()} | error. lookup_pid(Id) when is_integer(Id) -> case whereis(get_name(Id)) of undefined -> error; Pid -> {ok, Pid} end. -spec get_name(Id :: integer()) -> atom(). get_name(Id) when is_integer(Id) -> list_to_atom("sdlan_network:" ++ integer_to_list(Id)). -spec get_network_id(Pid :: pid()) -> {ok, NetworkId :: integer()}. get_network_id(Pid) when is_pid(Pid) -> gen_server:call(Pid, get_network_id). -spec attach(Pid :: pid(), ChannelPid :: pid(), Transport :: module(), ClientId :: binary(), Mac :: binary(), Ip :: integer(), Hostname :: binary()) -> {ok, Algorithm :: binary(), Key :: binary(), RegionId :: integer(), SessionToken :: binary()}. attach(Pid, ChannelPid, Transport, ClientId, Mac, Ip, Hostname) when is_pid(Pid), is_pid(ChannelPid), is_atom(Transport), is_binary(ClientId), is_binary(Mac), is_integer(Ip), is_binary(Hostname) -> gen_server:call(Pid, {attach, ChannelPid, Transport, ClientId, Mac, Ip, Hostname}). -spec unregister(Pid :: pid(), ClientId :: binary(), Mac :: binary()) -> ok. unregister(Pid, ClientId, Mac) when is_pid(Pid), is_binary(ClientId), is_binary(Mac) -> gen_server:cast(Pid, {unregister, ClientId, Mac}). -spec peer_info(Pid :: pid(), SrcMac :: binary(), DstMac :: binary()) -> error | {ok, {NatPeer :: {Ip :: inet:ip4_address(), Port :: integer()}, NatType :: integer()}, V6Info :: undefined | #'SDLV6Info'{}}. peer_info(Pid, SrcMac, DstMac) when is_pid(Pid), is_binary(SrcMac), is_binary(DstMac) -> gen_server:call(Pid, {peer_info, SrcMac, DstMac}). -spec arp_request(Pid :: pid(), TargetIp :: integer()) -> error | {ok, Mac :: binary()}. arp_request(Pid, TargetIp) when is_pid(Pid), is_integer(TargetIp) -> gen_server:call(Pid, {arp_request, TargetIp}). -spec command(Pid :: pid(), ReceiverPid :: pid(), ClientId :: binary(), {Tag :: atom(), SubCommand :: any()}) -> {error, Reason :: binary()} | {ok, Ref :: reference()}. command(Pid, ReceiverPid, ClientId, SubCommand) when is_pid(Pid), is_pid(ReceiverPid), is_binary(ClientId) -> gen_server:call(Pid, {command, ReceiverPid, ClientId, SubCommand}). -spec wait_command_ack(Ref :: reference(), Timeout :: integer()) -> {error, timeout} | {ok, CommandAck :: #'SDLCommandAck'{}}. wait_command_ack(Ref, Timeout) when is_reference(Ref), is_integer(Timeout) -> receive {quic_command_ack, Ref, CommandAck} -> {ok, CommandAck} after Timeout -> {error, timeout} end. -spec forward_by_ets(NetworkId :: integer(), Sock :: any(), SrcMac :: binary(), DstMac :: binary(), Packet :: binary()) -> {ok, ForwardBytes :: integer()} | {error, Reason :: any()}. forward_by_ets(NetworkId, Sock, SrcMac, DstMac, Packet) when is_integer(NetworkId), is_binary(SrcMac), is_binary(DstMac), is_binary(Packet) -> case endpoint_existing_table_name(NetworkId) of {ok, Table} -> case lookup_endpoint(Table, SrcMac) of #endpoint{} -> case sdlan_util:is_broadcast_mac(DstMac) orelse sdlan_util:is_multicast_mac(DstMac) of true -> forward_broadcast_by_ets(NetworkId, Sock, SrcMac, DstMac, Packet); false -> forward_unicast_by_ets(NetworkId, Sock, SrcMac, DstMac, Packet) end; undefined -> logger:debug("[sdlan_network] networkd_id: ~p, src_mac: ~p, dst_mac: ~p, forward discard, src not found", [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), {error, src_not_found} end; error -> {error, table_not_found} end. %% 更新ip地址对应的nat关系 -spec update_hole(Pid :: pid(), SessionToken :: binary(), ClientId :: binary(), Mac :: binary(), Peer :: tuple(), NatType :: integer(), V6Info :: undefined | #'SDLV6Info'{}) -> ok. update_hole(Pid, SessionToken, ClientId, Mac, Peer, NatType, V6Info) when is_pid(Pid), is_binary(ClientId), is_binary(Mac), is_integer(NatType) -> gen_server:cast(Pid, {update_hole, SessionToken, ClientId, Mac, Peer, NatType, V6Info}). -spec disable_client(Pid :: pid(), ClientId :: binary()) -> ok | error. disable_client(Pid, ClientId) when is_pid(Pid), is_binary(ClientId) -> gen_server:call(Pid, {disable_client, ClientId}). -spec acl_changed(Pid :: pid(), ClientId :: binary()) -> ok | error. acl_changed(Pid, ClientId) when is_pid(Pid), is_binary(ClientId) -> gen_server:call(Pid, {acl_changed, ClientId}). -spec get_channel(Pid :: pid(), ClientId :: binary()) -> error | {ok, ChannelPid :: pid()}. get_channel(Pid, ClientId) when is_pid(Pid), is_binary(ClientId) -> gen_server:call(Pid, {get_channel, ClientId}). -spec debug_info(Pid :: pid()) -> map(). debug_info(Pid) when is_pid(Pid) -> gen_server:call(Pid, debug_info). %% @doc Spawns the server and registers the local name (unique) -spec(start_link(Name :: atom(), NetworkData :: tuple()) -> {ok, Pid :: pid()} | ignore | {error, Reason :: term()}). start_link(Name, NetworkData = {network, Id, _NetworkName, _Domain, _IpAddr, _MaskLen, _OwnerId, _Algorithm}) when is_atom(Name), is_integer(Id) -> gen_server:start_link({local, Name}, ?MODULE, [NetworkData], []). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== %% @private %% @doc Initializes the server -spec(init(Args :: term()) -> {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} | {stop, Reason :: term()} | ignore). init([{network, Id, Name, Domain, IpAddr, MaskLen, OwnerId, Algorithm0}]) -> erlang:process_flag(trap_exit, true), %% 限流key ThrottleKey = list_to_atom("network_throttle:" ++ integer_to_list(Id)), %% 绑定到资源协调器 sdlan_network_coordinator:attach(self(), ThrottleKey), sdlan_domain_regedit:insert(Domain), %% 处理加密算法 Algorithm = normalization_algorithm(Algorithm0), Key = gen_key(Algorithm), EndpointTable = new_endpoint_table(Id), {ok, #state{network_id = Id, name = Name, domain = Domain, ipaddr = IpAddr, algorithm = Algorithm, owner_id = OwnerId, mask_len = MaskLen, key = Key, throttle_key = ThrottleKey, endpoint_table = EndpointTable}}. %% @private %% @doc Handling call messages -spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()}, State :: #state{}) -> {reply, Reply :: term(), NewState :: #state{}} | {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} | {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} | {stop, Reason :: term(), NewState :: #state{}}). %% 给客户端分配ip地址 handle_call({attach, ChannelPid, Transport, ClientId, Mac, Ip, Hostname}, _From, State = #state{network_id = NetworkId, domain = Domain, algorithm = Algorithm, key = Key, endpoint_table = EndpointTable}) -> %% 分配ip地址的时候,以mac地址为唯一基准 logger:debug("[sdlan_network] alloc_ip, network_id: ~p, client_id: ~p, mac: ~p, ip_addr: ~p", [NetworkId, ClientId, sdlan_util:format_mac(Mac), sdlan_util:int_to_ipv4(Ip)]), %% 添加域名->ip的映射关系 sdlan_hostname_regedit:insert(Hostname, Domain, Ip), %% mac对应的Endpoint存在,并且对应的ip变了,需要通知端上清理arp maybe_nat_changed(ChannelPid, Mac, Ip, EndpointTable), ChannelRef = monitor(process, ChannelPid), SessionToken = gen_session_token(), Endpoint = #endpoint{ channel_pid = ChannelPid, channel_ref = ChannelRef, transport = Transport, client_id = ClientId, mac = Mac, ip = Ip, hostname = Hostname, session_token = SessionToken, last_seen = erlang:monotonic_time(second) }, insert_endpoint(EndpointTable, Mac, Endpoint), %% 生成对应的分区id RegionId = gen_region_id(Ip), {reply, {ok, Algorithm, Key, RegionId, SessionToken}, State}; %% client设置为禁止状态,不允许重连 handle_call({disable_client, ClientId}, _From, State = #state{endpoint_table = EndpointTable}) -> MatchSpec = ets:fun2ms(fun(Object = {_Mac, #endpoint{client_id = ClientId0}}) when ClientId0 =:= ClientId -> Object end), case select_endpoint(EndpointTable, MatchSpec) of {ok, Mac, Endpoint} -> cleanup_endpoint(Endpoint, undefined, disabled), delete_endpoint(EndpointTable, Mac), {reply, ok, State}; error -> {reply, ok, State} end; handle_call({get_channel, ClientId}, _From, State = #state{endpoint_table = EndpointTable}) -> MatchSpec = ets:fun2ms(fun(Object = {_Mac, #endpoint{client_id = ClientId0}}) when ClientId0 =:= ClientId -> Object end), case select_endpoint(EndpointTable, MatchSpec) of {ok, _, #endpoint{channel_pid = ChannelPid}} -> {reply, {ok, ChannelPid}, State}; error -> {reply, error, State} end; handle_call(get_network_id, _From, State = #state{network_id = NetworkId}) -> {reply, {ok, NetworkId}, State}; %% 网络存在的nat_peer信息 handle_call({peer_info, SrcMac, DstMac}, _From, State = #state{endpoint_table = EndpointTable}) -> case lookup_endpoint(EndpointTable, DstMac) of #endpoint{channel_pid = DstChannelPid, transport = Transport, hole = #hole{peer = DstNatPeer, nat_type = DstNatType}, v6_info = DstV6Info} -> %% 让目标服务器发送sendRegister事件(2024-06-25 新增,提高打洞的成功率) maybe #endpoint{hole = #hole{peer = {SrcNatIp, SrcNatPort}, nat_type = SrcNatType}, v6_info = SrcV6Info} ?= lookup_endpoint(EndpointTable, SrcMac), RegisterEvent = sdlan_pb:encode_msg(#'SDLEvent' { event = {send_register, #'SDLEvent.SendRegister'{ dst_mac = SrcMac, nat_ip = sdlan_util:ipv4_to_int(SrcNatIp), nat_type = SrcNatType, nat_port = SrcNatPort, v6_info = SrcV6Info }} }), logger:debug("Event: send_register, for peer_info"), Transport:send_event(DstChannelPid, RegisterEvent) end, {reply, {ok, {DstNatPeer, DstNatType}, DstV6Info}, State}; _ -> {reply, error, State} end; %% arp查询 handle_call({arp_request, TargetIp}, _From, State = #state{endpoint_table = EndpointTable}) -> MatchSpec = ets:fun2ms(fun(Object = {_Mac, #endpoint{ip = Ip}}) when Ip =:= TargetIp -> Object end), case select_endpoint(EndpointTable, MatchSpec) of error -> {reply, error, State}; {ok, Mac, _} -> {reply, {ok, Mac}, State} end; %% 发送命令 handle_call({command, ReceiverPid, ClientId, SubCommand}, _From, State = #state{endpoint_table = EndpointTable}) -> MatchSpec = ets:fun2ms(fun(Object = {_Mac, #endpoint{client_id = ClientId0}}) when ClientId0 =:= ClientId -> Object end), case select_endpoint(EndpointTable, MatchSpec) of {ok, _Mac, #endpoint{channel_pid = ChannelPid, transport = Transport}} -> Ref = make_ref(), Transport:command(ChannelPid, Ref, ReceiverPid, SubCommand), {reply, {ok, Ref}, State}; error -> {reply, {error, <<"目标Node不在线"/utf8>>}, State} end; %% 触发acl改变 handle_call({acl_changed, ReceiverPid, ClientId, SubCommand}, _From, State = #state{endpoint_table = EndpointTable}) -> MatchSpec = ets:fun2ms(fun(Object = {_Mac, #endpoint{client_id = ClientId0}}) when ClientId0 =:= ClientId -> Object end), case select_endpoint(EndpointTable, MatchSpec) of {ok, _Mac, #endpoint{channel_pid = ChannelPid, transport = Transport}} -> Ref = make_ref(), Transport:command(ChannelPid, Ref, ReceiverPid, SubCommand), {reply, {ok, Ref}, State}; error -> {reply, {error, <<"目标Node不在线"/utf8>>}, State} end; handle_call(debug_info, _From, State = #state{network_id = NetworkId, ipaddr = IpAddr, mask_len = MaskLen, owner_id = OwnerId, endpoint_table = EndpointTable}) -> Reply = #{ <<"network_id">> => NetworkId, <<"ipaddr">> => IpAddr, <<"mask_len">> => MaskLen, <<"owner_id">> => OwnerId, <<"metrics">> => network_metrics(State), <<"used_ips">> => lists:map(fun format_endpoint/1, list_endpoints(EndpointTable)) }, {reply, Reply, State}. %% @private %% @doc Handling cast messages -spec(handle_cast(Request :: term(), State :: #state{}) -> {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). %% 删除ip的占用并关闭channel handle_cast({unregister, ClientId, Mac}, State = #state{network_id = NetworkId, endpoint_table = EndpointTable}) -> logger:debug("[sdlan_network] networkd_id: ~p, unregister client_id: ~p, Mac: ~p", [NetworkId, ClientId, sdlan_util:format_mac(Mac)]), case lookup_endpoint(EndpointTable, Mac) of Endpoint = #endpoint{client_id = ClientId} -> cleanup_endpoint(Endpoint, Endpoint#endpoint.channel_pid, unregister), delete_endpoint(EndpointTable, Mac), {noreply, State}; undefined -> {noreply, State} end; %% 需要判断,client是属于当前网络的 handle_cast({update_hole, SessionToken, ClientId, Mac, Peer, NatType, V6Info}, State = #state{endpoint_table = EndpointTable}) -> case lookup_endpoint(EndpointTable, Mac) of %% ClientId =:= ClientId0, SessionToken =:= SessionToken0 Endpoint0 = #endpoint{ip = Ip, client_id = ClientId, hole = OldHole, session_token = SessionToken} -> NHole = #hole{peer = Peer, nat_type = NatType}, maybe true ?= not same_hole(OldHole, NHole), NatChangedEvent = sdlan_pb:encode_msg(#'SDLEvent' { event = {nat_changed, #'SDLEvent.NatChanged'{ mac = Mac, ip = Ip }} }), logger:debug("[sdlan_network] Event: nat_changed, update_hole, client_id: ~p(~p), hole changed", [ClientId, Ip]), broadcast(fun(#endpoint{channel_pid = ChannelPid, transport = Transport}) -> Transport:send_event(ChannelPid, NatChangedEvent) end, [Mac], EndpointTable) end, NEndpoint = Endpoint0#endpoint{hole = NHole, v6_info = V6Info, last_seen = erlang:monotonic_time(second)}, insert_endpoint(EndpointTable, Mac, NEndpoint), logger:debug("[sdlan_network] mac: ~p, ip: ~p, endpoint is: ~p", [Mac, Ip, NEndpoint]), {noreply, State}; _ -> {noreply, State} end. %% @private %% @doc Handling all non call/cast messages -spec(handle_info(Info :: timeout() | term(), State :: #state{}) -> {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). %% Channel进程退出, hole里面的数据也需要清理 handle_info({'DOWN', _MRef, process, ChannelPid, Reason}, State = #state{network_id = NetworkId, endpoint_table = EndpointTable}) -> logger:notice("[sdlan_network] network_id: ~p, channel_pid: ~p, close with reason: ~p", [NetworkId, ChannelPid, Reason]), remove_channel_endpoints(ChannelPid, EndpointTable), {noreply, State}; handle_info(Info, State) -> logger:debug("[sdlan_network] get unknown info: ~p", [Info]), {noreply, State}. %% @private %% @doc This function is called by a gen_server when it is about to %% terminate. It should be the opposite of Module:init/1 and do any %% necessary cleaning up. When it returns, the gen_server terminates %% with Reason. The return value is ignored. -spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()), State :: #state{}) -> ok). terminate(Reason, #state{network_id = NetworkId, endpoint_table = EndpointTable}) -> broadcast(fun(#endpoint{channel_pid = ChannelPid, transport = Transport}) -> case is_pid(ChannelPid) andalso is_process_alive(ChannelPid) of true -> NetworkShutdownEvent = sdlan_pb:encode_msg(#'SDLEvent'{ event = {shutdown, #'SDLEvent.NetworkShutdown'{ message = <<"Network shutdown">> }} }), logger:debug("[sdlan_network] Event: shutdown"), Transport:send_event(ChannelPid, NetworkShutdownEvent), Transport:stop(ChannelPid, normal); false -> ok end end, [], EndpointTable), logger:debug("[sdlan_network] network: ~p, will terminate with reason: ~p", [NetworkId, Reason]), ok. %% @private %% @doc Convert process state when code is changed -spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{}, Extra :: term()) -> {ok, NewState :: #state{}} | {error, Reason :: term()}). code_change(_OldVsn, State = #state{}, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== -spec limiting_check(ThrottleKey :: any()) -> pass | denied. limiting_check(ThrottleKey) -> case throttle:check(sdlan_network, ThrottleKey) of {ok, _RestCount, _LeftToReset} -> pass; {limit_exceeded, 0, _LeftToReset} -> %% 尝试获取其他网络是否有让渡的资源 case sdlan_network_coordinator:checkout() of ok -> pass; error -> denied end end. -spec endpoint_table_name(NetworkId :: integer()) -> atom(). endpoint_table_name(NetworkId) when is_integer(NetworkId) -> list_to_atom(lists:concat(["sdlan_network_endpoint:", NetworkId])). -spec endpoint_existing_table_name(NetworkId :: integer()) -> {ok, atom()} | error. endpoint_existing_table_name(NetworkId) when is_integer(NetworkId) -> try Atom = list_to_existing_atom(lists:concat(["sdlan_network_endpoint:", NetworkId])), {ok, Atom} catch error:_ -> error end. -spec throttle_key(NetworkId :: integer()) -> atom(). throttle_key(NetworkId) when is_integer(NetworkId) -> list_to_atom("network_throttle:" ++ integer_to_list(NetworkId)). -spec new_endpoint_table(NetworkId :: integer()) -> ets:tid(). new_endpoint_table(NetworkId) when is_integer(NetworkId) -> ets:new(endpoint_table_name(NetworkId), [ named_table, protected, set, {read_concurrency, true}, {write_concurrency, true} ]). -spec insert_endpoint(Table :: ets:tid(), Mac :: binary(), Endpoint :: #endpoint{}) -> true. insert_endpoint(Table, Mac, Endpoint = #endpoint{}) when is_binary(Mac) -> ets:insert(Table, {Mac, Endpoint}). -spec delete_endpoint(Table :: ets:tid(), Mac :: binary()) -> true. delete_endpoint(Table, Mac) when is_binary(Mac) -> ets:delete(Table, Mac). -spec lookup_endpoint(atom() | ets:tid(), Mac :: binary()) -> #endpoint{} | undefined. lookup_endpoint(Table, Mac) when is_binary(Mac) -> try ets:lookup(Table, Mac) of [{Mac, Endpoint = #endpoint{}}] -> Endpoint; [] -> undefined catch error:_ -> undefined end. -spec list_endpoints(atom() | ets:tid()) -> [{binary(), #endpoint{}}]. list_endpoints(Table) -> try ets:tab2list(Table) of Endpoints when is_list(Endpoints) -> Endpoints catch error:_ -> [] end. -spec remove_channel_endpoints(ChannelPid :: pid(), Table :: ets:tid()) -> ok. remove_channel_endpoints(ChannelPid, Table) when is_pid(ChannelPid) -> lists:foreach(fun ({Mac, #endpoint{channel_pid = ChannelPid0}}) when ChannelPid =:= ChannelPid0 -> delete_endpoint(Table, Mac); (_) -> ok end, list_endpoints(Table)). -spec forward_unicast_by_ets(NetworkId :: integer(), Sock :: any(), SrcMac :: binary(), DstMac :: binary(), Packet :: binary()) -> {ok, integer()} | {error, any()}. forward_unicast_by_ets(NetworkId, Sock, SrcMac, DstMac, Packet) -> PacketBytes = byte_size(Packet), case endpoint_existing_table_name(NetworkId) of {ok, Table} -> case lookup_endpoint(Table, DstMac) of #endpoint{hole = #hole{peer = Peer = {NatIp, NatPort}}} -> case limiting_check(throttle_key(NetworkId)) of pass -> logger:debug("[sdlan_network] forward data by ets networkd_id: ~p, src_mac: ~p, dst_mac: ~p, hole: ~p", [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac), Peer]), gen_udp:send(Sock, NatIp, NatPort, Packet), {ok, PacketBytes}; denied -> logger:notice("[sdlan_network] networkd_id: ~p, src_mac: ~p, dst_mac: ~p, rate limited, discard", [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), {error, rate_limited} end; #endpoint{} -> logger:debug("[sdlan_network] networkd_id: ~p, src_mac: ~p, dst_mac: ~p, hole not found", [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), {error, hole_not_found}; undefined -> logger:debug("[sdlan_network] networkd_id: ~p, src_mac: ~p, dst_mac: ~p not found", [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), {error, dst_not_found} end; error -> {error, table_not_found} end. -spec forward_broadcast_by_ets(NetworkId :: integer(), Sock :: any(), SrcMac :: binary(), DstMac :: binary(), Packet :: binary()) -> {ok, integer()}. forward_broadcast_by_ets(NetworkId, Sock, SrcMac, DstMac, Packet) -> Table = endpoint_table_name(NetworkId), lists:foreach(fun ({Mac, #endpoint{hole = #hole{peer = {NatIp, NatPort}}}}) when Mac =/= SrcMac -> gen_udp:send(Sock, NatIp, NatPort, Packet); (_) -> ok end, list_endpoints(Table)), logger:debug("[sdlan_network] broadcast data by ets networkd_id: ~p, src_mac: ~p, dst_mac: ~p", [NetworkId, sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac)]), {ok, byte_size(Packet)}. maybe_nat_changed(ChannelPid, Mac, Ip, EndpointTable) -> OldEndpoint = lookup_endpoint(EndpointTable, Mac), %% mac对应的Endpoint存在,并且对应的ip变了,需要通知端上清理arp maybe #endpoint{ip = OldIp} ?= OldEndpoint, true ?= OldIp =/= Ip, Event = sdlan_pb:encode_msg(#'SDLEvent'{ event = {nat_changed, #'SDLEvent.NatChanged' { mac = Mac, ip = Ip }} }), logger:debug("Event: nat_changed, for attach"), broadcast(fun(#endpoint{channel_pid = ChannelPid0, transport = Transport}) -> Transport:send_event(ChannelPid0, Event) end, [Mac], EndpointTable) end, %% 重复attach需要清理之前的绑定信息;即使IP未变化,也不能保留旧channel。 cleanup_endpoint(OldEndpoint, ChannelPid, rebind). cleanup_endpoint(undefined, _KeepChannelPid, _Reason) -> ok; cleanup_endpoint(#endpoint{channel_ref = ChannelRef, channel_pid = ChannelPid, transport = Transport}, KeepChannelPid, Reason) -> is_reference(ChannelRef) andalso erlang:demonitor(ChannelRef, [flush]), case should_stop_channel(ChannelPid, KeepChannelPid) of true -> catch Transport:stop(ChannelPid, Reason), ok; false -> ok end. should_stop_channel(ChannelPid, KeepChannelPid) when is_pid(ChannelPid), ChannelPid =/= KeepChannelPid -> is_process_alive(ChannelPid); should_stop_channel(_, _) -> false. -spec broadcast(Fun :: fun((#endpoint{}) -> term()), ExcludeMacs :: [binary()], Table :: ets:tid()) -> ok. broadcast(Fun, ExcludeMacs, Table) when is_function(Fun, 1), is_list(ExcludeMacs) -> lists:foreach(fun({Mac, Endpoint}) -> case lists:member(Mac, ExcludeMacs) of true -> ok; false -> Fun(Endpoint) end end, list_endpoints(Table)). -spec format_endpoint({Mac :: binary(), Host :: #endpoint{}}) -> map(). format_endpoint({Mac, #endpoint{client_id = ClientId, ip = Ip, hole = #hole{peer = {NatIp, NatPort}, nat_type = NatType}, v6_info = V6Info}}) -> HoleMap = #{ nat_ip => NatIp, nat_port => NatPort, nat_type => NatType }, V6InfoMap = case V6Info of undefined -> #{}; #'SDLV6Info'{v6 = V6, port = V6Port} -> #{v6 => V6, port => V6Port} end, #{ client_id => ClientId, mac => sdlan_util:format_mac(Mac), ip => sdlan_util:int_to_ipv4(Ip), hole_map => HoleMap, v6_info => V6InfoMap }; format_endpoint({Mac, #endpoint{client_id = ClientId, ip = Ip, hole = undefined, v6_info = V6Info}}) -> V6InfoMap = case V6Info of undefined -> #{}; #'SDLV6Info'{v6 = V6, port = V6Port} -> #{v6 => V6, port => V6Port} end, #{ client_id => ClientId, mac => sdlan_util:format_mac(Mac), ip => sdlan_util:int_to_ipv4(Ip), hole_map => undefined, v6_info => V6InfoMap }. network_metrics(#state{network_id = NetworkId, endpoint_table = EndpointTable}) -> ProcInfo = maps:from_list(process_info(self(), [message_queue_len, memory, reductions])), ProcInfo#{ network_id => NetworkId, endpoint_count => endpoint_count(EndpointTable), channel_metrics => channel_metrics(EndpointTable) }. endpoint_count(Table) -> case catch ets:info(Table, size) of Size when is_integer(Size) -> Size; _ -> 0 end. channel_metrics(Table) -> lists:foldl(fun({_Mac, #endpoint{channel_pid = ChannelPid}}, Acc) -> case is_pid(ChannelPid) andalso is_process_alive(ChannelPid) of true -> ProcInfo = maps:from_list(process_info(ChannelPid, [message_queue_len, memory])), [ProcInfo#{pid => ChannelPid} | Acc]; false -> Acc end end, [], list_endpoints(Table)). -spec select_endpoint(Table :: ets:tid(), MatchSpec :: ets:match_spec()) -> error | {ok, Mac :: binary(), Endpoint :: #endpoint{}}. select_endpoint(Table, MatchSpec) -> case catch ets:select(Table, MatchSpec, 1) of {[], _Continuation} -> error; {[{Mac, Endpoint = #endpoint{}}], _Continuation} -> {ok, Mac, Endpoint}; '$end_of_table' -> error; {'EXIT', _} -> error end. -spec same_hole(Hole :: #hole{}, Hole :: #hole{}) -> boolean(). same_hole(#hole{peer = OldPeer, nat_type = OldNatType}, #hole{peer = Peer, nat_type = NatType}) when OldPeer =:= Peer, OldNatType =:= NatType -> true; same_hole(_, _) -> false. -spec gen_session_token() -> binary(). gen_session_token() -> Bytes = crypto:strong_rand_bytes(32), base64:encode(Bytes). -spec normalization_algorithm(any()) -> binary(). normalization_algorithm(<<"aes">>) -> <<"aes">>; normalization_algorithm(<<"chacha20">>) -> <<"chacha20">>; normalization_algorithm(_) -> <<"chacha20">>. -spec gen_key(Algorithm :: binary()) -> Key :: binary(). gen_key(<<"aes">>) -> sdlan_util:rand_byte(32); gen_key(<<"chacha20">>) -> sdlan_util:rand_byte(32). -spec gen_region_id(IpInt :: integer()) -> integer(). gen_region_id(IpInt) -> %% 把整数 IP 转成字符串 IpStr = integer_to_list(IpInt), %% 拼接盐 FullStr = "salt_fG7xQp2BzH9L" ++ IpStr, time33(FullStr, 5381). %% 核心Time33算法 -spec time33(string(), integer()) -> integer(). time33([], Hash) -> Hash band 16#FFFFFFFF; % 32位 time33([C|Rest], Hash) -> %% hash = hash * 33 + char NewHash = ((Hash bsl 5) + Hash) + C, time33(Rest, NewHash).