diff --git a/.credo.exs b/.credo.exs index 7dceb12..dbebfee 100644 --- a/.credo.exs +++ b/.credo.exs @@ -122,7 +122,7 @@ # {Credo.Check.Refactor.Apply, []}, {Credo.Check.Refactor.CondStatements, []}, - {Credo.Check.Refactor.CyclomaticComplexity, [max_complexity: 12]}, + {Credo.Check.Refactor.CyclomaticComplexity, [max_complexity: 11]}, {Credo.Check.Refactor.FunctionArity, []}, {Credo.Check.Refactor.LongQuoteBlocks, []}, {Credo.Check.Refactor.MatchInCondition, []}, diff --git a/README.md b/README.md index 8b87a2d..4ea3d54 100644 --- a/README.md +++ b/README.md @@ -30,7 +30,7 @@ as WebRTC multiplexes traffic on a single socket but PRs are welcomed ```elixir def deps do [ - {:ex_ice, "~> 0.15.0"} + {:ex_ice, "~> 0.16.0"} ] end ``` diff --git a/lib/ex_ice/priv/ice_agent.ex b/lib/ex_ice/priv/ice_agent.ex index 82a73ef..9235e34 100644 --- a/lib/ex_ice/priv/ice_agent.ex +++ b/lib/ex_ice/priv/ice_agent.ex @@ -943,63 +943,92 @@ defmodule ExICE.Priv.ICEAgent do def handle_ex_turn_msg(ice_agent, client_ref, msg) do tr_id_tr = find_gathering_transaction(ice_agent.gathering_transactions, client_ref) - cand = find_relay_cand_by_client(Map.values(ice_agent.local_cands), client_ref) case {tr_id_tr, cand} do - {nil, nil} -> - ice_agent + {nil, nil} -> ice_agent + {{tr_id, tr}, nil} -> handle_ex_turn_msg_for_gathering_tx(ice_agent, tr_id, tr, msg) + {nil, cand} -> handle_ex_turn_msg_for_relay_cand(ice_agent, cand, msg) + end + end - {{tr_id, tr}, nil} -> - case ExTURN.Client.handle_message(tr.client, msg) do - {:ok, client} -> - tr = %{tr | client: client} - put_in(ice_agent.gathering_transactions[tr_id], tr) + defp handle_ex_turn_msg_for_gathering_tx(ice_agent, tr_id, tr, msg) do + case ExTURN.Client.handle_message(tr.client, msg) do + {:ok, client} -> + put_in(ice_agent.gathering_transactions[tr_id], %{tr | client: client}) - {:send, dst, data, client} -> - tr = %{tr | client: client} - :ok = ice_agent.transport_module.send(tr.socket, dst, data) - put_in(ice_agent.gathering_transactions[tr_id], tr) + {:send, dst, data, client} -> + send_on_gathering_tx(ice_agent, tr_id, %{tr | client: client}, dst, data) - {:error, _reason, _client} -> - {_, ice_agent} = pop_in(ice_agent.gathering_transactions[tr_id]) - update_gathering_state(ice_agent) - end + {:error, _reason, _client} -> + drop_gathering_transaction(ice_agent, tr_id) + end + end - {nil, cand} -> - case ExTURN.Client.handle_message(cand.client, msg) do - {:ok, client} -> - cand = %{cand | client: client} - put_in(ice_agent.local_cands[cand.base.id], cand) - - {:send, dst, data, client} -> - cand = %{cand | client: client} - ice_agent = put_in(ice_agent.local_cands[cand.base.id], cand) - # we can't use do_send here as it will try to create permission for the turn address - :ok = ice_agent.transport_module.send(cand.base.socket, dst, data) - ice_agent + defp send_on_gathering_tx(ice_agent, tr_id, tr, dst, data) do + case ice_agent.transport_module.send(tr.socket, dst, data) do + :ok -> + put_in(ice_agent.gathering_transactions[tr_id], tr) + + {:error, _reason} -> + drop_gathering_transaction(ice_agent, tr_id) + end + end + + defp drop_gathering_transaction(ice_agent, tr_id) do + {_, ice_agent} = pop_in(ice_agent.gathering_transactions[tr_id]) + update_gathering_state(ice_agent) + end - {:permission_expired, _ip, client} -> - cand = %{cand | client: client} - put_in(ice_agent.local_cands[cand.base.id], cand) + defp handle_ex_turn_msg_for_relay_cand(ice_agent, cand, msg) do + case ExTURN.Client.handle_message(cand.client, msg) do + {:ok, client} -> + update_relay_cand_client(ice_agent, cand, client) - {:channel_expired, _addr, client} -> - cand = %{cand | client: client} - put_in(ice_agent.local_cands[cand.base.id], cand) + {:send, dst, data, client} -> + send_on_relay_cand(ice_agent, cand, client, dst, data) - {:error, _reason, client} -> - Logger.debug(""" - Couldn't handle TURN message on candidate: #{inspect(cand)}. \ - Closing candidate.\ - """) + {:permission_expired, _ip, client} -> + update_relay_cand_client(ice_agent, cand, client) - cand = %{cand | client: client} - ice_agent = put_in(ice_agent.local_cands[cand.base.id], cand) - close_candidate(ice_agent, cand) - end + {:channel_expired, _addr, client} -> + update_relay_cand_client(ice_agent, cand, client) + + {:error, _reason, client} -> + Logger.debug(""" + Couldn't handle TURN message on candidate: #{inspect(cand)}. \ + Closing candidate.\ + """) + + ice_agent + |> update_relay_cand_client(cand, client) + |> close_candidate(cand) end end + # we can't use do_send here as it will try to create permission for the turn address + defp send_on_relay_cand(ice_agent, cand, client, dst, data) do + cand = %{cand | client: client} + ice_agent = put_in(ice_agent.local_cands[cand.base.id], cand) + + case ice_agent.transport_module.send(cand.base.socket, dst, data) do + :ok -> + ice_agent + + {:error, _reason} -> + Logger.debug(""" + Couldn't send TURN message on candidate: #{inspect(cand)}. \ + Closing candidate.\ + """) + + close_candidate(ice_agent, cand) + end + end + + defp update_relay_cand_client(ice_agent, cand, client) do + put_in(ice_agent.local_cands[cand.base.id], %{cand | client: client}) + end + @spec close(t()) :: t() def close(%__MODULE__{state: :closed} = ice_agent) do ice_agent @@ -1283,8 +1312,16 @@ defmodule ExICE.Priv.ICEAgent do {:send, turn_addr, data, client} -> tr = %{tr | client: client} - :ok = ice_agent.transport_module.send(tr.socket, turn_addr, data) - put_in(ice_agent.gathering_transactions[tr_id], tr) + + case ice_agent.transport_module.send(tr.socket, turn_addr, data) do + :ok -> + put_in(ice_agent.gathering_transactions[tr_id], tr) + + {:error, _reason} -> + Logger.debug("Failed to create TURN allocation.") + {_, ice_agent} = pop_in(ice_agent.gathering_transactions[tr_id]) + ice_agent + end {:error, _reason, _client} -> Logger.debug("Failed to create TURN allocation.") @@ -2019,16 +2056,27 @@ defmodule ExICE.Priv.ICEAgent do ice_agent end - defp handle_keepalive_response( + defp handle_keepalive_response(ice_agent, local_cand, src_ip, src_port, msg) do + {pair_id, ice_agent} = pop_in(ice_agent.keepalives[msg.transaction_id]) + + case Map.fetch(ice_agent.checklist, pair_id) do + {:ok, %CandidatePair{} = pair} -> + handle_keepalive_response_on_pair(ice_agent, local_cand, src_ip, src_port, msg, pair) + + :error -> + Logger.debug("Ignoring keepalive response for pruned pair #{inspect(pair_id)}") + ice_agent + end + end + + defp handle_keepalive_response_on_pair( ice_agent, local_cand, src_ip, src_port, - %Message{type: %Type{class: :success_response}} = msg + %Message{type: %Type{class: :success_response}} = msg, + pair ) do - {pair_id, ice_agent} = pop_in(ice_agent.keepalives[msg.transaction_id]) - %CandidatePair{} = pair = Map.fetch!(ice_agent.checklist, pair_id) - with true <- symmetric?(ice_agent, local_cand.base.socket, {src_ip, src_port}, pair), :ok <- authenticate_msg(msg, ice_agent.remote_pwd) do Logger.debug("Received keepalive success response on: #{pair_info(ice_agent, pair)}") @@ -2045,7 +2093,7 @@ defmodule ExICE.Priv.ICEAgent do ka_local_cand = Map.fetch!(ice_agent.local_cands, pair.local_cand_id) ka_remote_cand = Map.fetch!(ice_agent.remote_cands, pair.remote_cand_id) - pair = %CandidatePair{ + pair = %{ pair | non_symmetric_responses_received: pair.non_symmetric_responses_received + 1 } @@ -2068,20 +2116,19 @@ defmodule ExICE.Priv.ICEAgent do Not refreshing last_seen time.\ """) - pair = %CandidatePair{pair | responses_received: pair.responses_received + 1} + pair = %{pair | responses_received: pair.responses_received + 1} put_in(ice_agent.checklist[pair.id], pair) end end - defp handle_keepalive_response( + defp handle_keepalive_response_on_pair( ice_agent, local_cand, src_ip, src_port, - %Message{type: %Type{class: :error_response}} = msg + %Message{type: %Type{class: :error_response}}, + pair ) do - {pair_id, ice_agent} = pop_in(ice_agent.keepalives[msg.transaction_id]) - %CandidatePair{} = pair = Map.fetch!(ice_agent.checklist, pair_id) pair = %{pair | responses_received: pair.responses_received + 1} ice_agent = put_in(ice_agent.checklist[pair.id], pair) diff --git a/mix.exs b/mix.exs index 7801057..11dcf03 100644 --- a/mix.exs +++ b/mix.exs @@ -1,7 +1,7 @@ defmodule ExICE.MixProject do use Mix.Project - @version "0.15.0" + @version "0.16.0" @source_url "https://github.com/elixir-webrtc/ex_ice" def project do diff --git a/test/priv/ice_agent_test.exs b/test/priv/ice_agent_test.exs index 0bd08dc..c400240 100644 --- a/test/priv/ice_agent_test.exs +++ b/test/priv/ice_agent_test.exs @@ -750,6 +750,62 @@ defmodule ExICE.Priv.ICEAgentTest do assert new_pair.last_seen == pair.last_seen assert new_pair.responses_received == pair.responses_received + 1 end + + test "success response for pruned pair", %{ice_agent: ice_agent} do + ice_agent = connect(ice_agent) + + [socket] = ice_agent.sockets + [remote_cand] = Map.values(ice_agent.remote_cands) + [pair] = Map.values(ice_agent.checklist) + + ice_agent = ICEAgent.handle_keepalive_timeout(ice_agent, pair.id) + assert packet = Transport.Mock.recv(socket) + assert {:ok, req} = ExSTUN.Message.decode(packet) + + resp = + binding_response( + req.transaction_id, + ice_agent.transport_module, + socket, + ice_agent.remote_pwd + ) + + {_, ice_agent} = pop_in(ice_agent.checklist[pair.id]) + assert ice_agent.checklist == %{} + + ice_agent = + ICEAgent.handle_udp(ice_agent, socket, remote_cand.address, remote_cand.port, resp) + + assert ice_agent.checklist == %{} + assert ice_agent.keepalives == %{} + end + + test "error response for pruned pair", %{ice_agent: ice_agent} do + ice_agent = connect(ice_agent) + + [socket] = ice_agent.sockets + [remote_cand] = Map.values(ice_agent.remote_cands) + [pair] = Map.values(ice_agent.checklist) + + ice_agent = ICEAgent.handle_keepalive_timeout(ice_agent, pair.id) + assert packet = Transport.Mock.recv(socket) + assert {:ok, req} = ExSTUN.Message.decode(packet) + + resp = + Message.new(req.transaction_id, %Type{class: :error_response, method: :binding}, [ + %ErrorCode{code: 400} + ]) + |> Message.encode() + + {_, ice_agent} = pop_in(ice_agent.checklist[pair.id]) + assert ice_agent.checklist == %{} + + ice_agent = + ICEAgent.handle_udp(ice_agent, socket, remote_cand.address, remote_cand.port, resp) + + assert ice_agent.checklist == %{} + assert ice_agent.keepalives == %{} + end end describe "incoming binding request" do @@ -2663,6 +2719,77 @@ defmodule ExICE.Priv.ICEAgentTest do refute Map.has_key?(cand.client.channel_addr, channel_number) end + test "send failure on gathering transaction via handle_ex_turn_msg", %{ + ice_agent: ice_agent + } do + [socket] = ice_agent.sockets + + ice_agent = ICEAgent.handle_ta_timeout(ice_agent) + req = read_allocate_request(socket) + resp = allocate_error_response(req.transaction_id) + + turn_tr_id = {socket, {@turn_ip, @turn_port}} + tr = Map.fetch!(ice_agent.gathering_transactions, turn_tr_id) + + Transport.Mock.fail_send(socket) + + ice_agent = + ICEAgent.handle_ex_turn_msg( + ice_agent, + tr.client.ref, + {:socket_data, @turn_ip, @turn_port, resp} + ) + + assert ice_agent.gathering_transactions == %{} + end + + test "send failure on gathering transaction via handle_udp", %{ice_agent: ice_agent} do + [socket] = ice_agent.sockets + + ice_agent = ICEAgent.handle_ta_timeout(ice_agent) + req = read_allocate_request(socket) + resp = allocate_error_response(req.transaction_id) + + Transport.Mock.fail_send(socket) + + ice_agent = ICEAgent.handle_udp(ice_agent, socket, @turn_ip, @turn_port, resp) + + assert ice_agent.gathering_transactions == %{} + + assert nil == + ice_agent.local_cands + |> Map.values() + |> Enum.find(&(&1.base.type == :relay)) + end + + test "send failure on relay candidate", %{ice_agent: ice_agent} do + [socket] = ice_agent.sockets + + ice_agent = ICEAgent.handle_ta_timeout(ice_agent) + req = read_allocate_request(socket) + resp = allocate_error_response(req.transaction_id) + ice_agent = ICEAgent.handle_udp(ice_agent, socket, @turn_ip, @turn_port, resp) + req = read_allocate_request(socket) + resp = allocate_success_response(req.transaction_id, ice_agent.transport_module, socket) + ice_agent = ICEAgent.handle_udp(ice_agent, socket, @turn_ip, @turn_port, resp) + + cand = ice_agent.local_cands |> Map.values() |> Enum.find(&(&1.base.type == :relay)) + assert %ExICE.Priv.Candidate.Relay{} = cand + refute cand.base.closed? + + Transport.Mock.fail_send(socket) + + ice_agent = + ICEAgent.handle_ex_turn_msg( + ice_agent, + cand.client.ref, + :refresh_alloc + ) + + updated_cand = ice_agent.local_cands[cand.base.id] + assert updated_cand.base.closed? + end + test "invalid TURN URL" do ice_agent = ICEAgent.new( diff --git a/test/support/transport/mock.ex b/test/support/transport/mock.ex index 8d71f0a..e81be3d 100644 --- a/test/support/transport/mock.ex +++ b/test/support/transport/mock.ex @@ -26,6 +26,20 @@ defmodule ExICE.Support.Transport.Mock do end end + @spec fail_send(ExICE.Priv.Transport.socket()) :: :ok + def fail_send(ref) do + [{^ref, socket}] = :ets.lookup(:transport_mock, ref) + :ets.insert(:transport_mock, {ref, Map.put(socket, :fail_send, true)}) + :ok + end + + @spec unfail_send(ExICE.Priv.Transport.socket()) :: :ok + def unfail_send(ref) do + [{^ref, socket}] = :ets.lookup(:transport_mock, ref) + :ets.insert(:transport_mock, {ref, Map.delete(socket, :fail_send)}) + :ok + end + @spec recv(ExICE.Priv.Transport.socket()) :: binary() | nil def recv(ref) do case :ets.lookup(:transport_mock, ref) do @@ -88,8 +102,13 @@ defmodule ExICE.Support.Transport.Mock do @impl true def send(ref, _dst, packet, _tp_opts \\ []) do [{^ref, %{state: :open} = socket}] = :ets.lookup(:transport_mock, ref) - :ets.insert(:transport_mock, {ref, %{socket | buf: socket.buf ++ [packet]}}) - :ok + + if socket[:fail_send] do + {:error, :closed} + else + :ets.insert(:transport_mock, {ref, %{socket | buf: socket.buf ++ [packet]}}) + :ok + end end @impl true