Ignore data frome stale sockets, timers, etc. Clean up state when connection is failed or completed

Author: mickel8

lib/ex_ice/priv/ice_agent.ex

@@ -220,6 +220,11 @@ defmodule ExICE.Priv.ICEAgent do
   end
 
   @spec set_remote_credentials(t(), binary(), binary()) :: t()
+  def set_remote_credentials(%__MODULE__{state: :failed} = ice_agent, _, _) do
+    Logger.debug("Tried to set remote credentials in failed state. ICE restart needed. Ignoring.")
+    ice_agent
+  end
+
   def set_remote_credentials(
         %__MODULE__{remote_ufrag: nil, remote_pwd: nil} = ice_agent,
         ufrag,
@@ -249,6 +254,11 @@ defmodule ExICE.Priv.ICEAgent do
   end
 
   @spec gather_candidates(t()) :: t()
+  def gather_candidates(%__MODULE__{state: :failed} = ice_agent) do
+    Logger.warning("Can't gather candidates in state failed. ICE restart needed. Ignoring.")
+    ice_agent
+  end
+
   def gather_candidates(%__MODULE__{gathering_state: :gathering} = ice_agent) do
     Logger.warning("Can't gather candidates. Gathering already in progress. Ignoring.")
     ice_agent
@@ -311,6 +321,12 @@ defmodule ExICE.Priv.ICEAgent do
   end
 
   @spec add_remote_candidate(t(), Candidate.t()) :: t()
+  def add_remote_candidate(%__MODULE__{state: :failed} = ice_agent, _) do
+    # Completed state will be cought by the next clause
+    Logger.debug("Can't add remote candidate in state failed. ICE restart needed. Ignoring.")
+    ice_agent
+  end
+
   def add_remote_candidate(%__MODULE__{eoc: true} = ice_agent, remote_cand) do
     Logger.warning("""
     Received remote candidate after end-of-candidates. Ignoring.
@@ -361,6 +377,11 @@ defmodule ExICE.Priv.ICEAgent do
   end
 
   @spec end_of_candidates(t()) :: t()
+  def end_of_candidates(%__MODULE__{state: :failed} = ice_agent) do
+    Logger.debug("Can't set end-of-candidates flag in state failed. Ignoring.")
+    ice_agent
+  end
+
   def end_of_candidates(%__MODULE__{role: :controlled} = ice_agent) do
     Logger.debug("Setting end-of-candidates flag.")
     ice_agent = %{ice_agent | eoc: true}
@@ -477,6 +498,11 @@ defmodule ExICE.Priv.ICEAgent do
   end
 
   @spec handle_eoc_timeout(t()) :: t()
+  def handle_eoc_timeout(%__MODULE__{state: :failed} = ice_agent) do
+    Logger.debug("EOC timer fired but we are in the failed state. Ignoring.")
+    %{ice_agent | eoc_timer: nil}
+  end
+
   def handle_eoc_timeout(%{eoc: true} = ice_agent) do
     Logger.debug("EOC timer fired but EOC flag is already set. Ignoring.")
     %{ice_agent | eoc_timer: nil}
@@ -489,6 +515,12 @@ defmodule ExICE.Priv.ICEAgent do
   end
 
   @spec handle_pair_timeout(t()) :: t()
+  def handle_pair_timeout(%__MODULE__{state: state} = ice_agent)
+      when state in [:failed, :completed] do
+    Logger.debug("Pair timeout timer fired in state: #{state}. Ignoring.")
+    ice_agent
+  end
+
   def handle_pair_timeout(ice_agent) do
     start_pair_timer()
 
@@ -564,8 +596,16 @@ defmodule ExICE.Priv.ICEAgent do
       {:ok, _pair} ->
         ice_agent
 
+      :error when ice_agent.state in [:failed, :completed] ->
+        Logger.warning("""
+        Received keepalive request for non-existant candidate pair but we are in state: #{ice_agent.state}. \
+        Ignoring.\
+        """)
+
+        ice_agent
+
       :error ->
-        Logger.warning("Received keepalive request for non-existant candidate pair")
+        Logger.warning("Received keepalive request for non-existant candidate pair. Ignoring.")
         ice_agent
     end
   end
@@ -577,6 +617,10 @@ defmodule ExICE.Priv.ICEAgent do
           :inet.port_number(),
           binary()
         ) :: t()
+  def handle_udp(%{state: :failed} = ice_agent, _socket, _src_ip, _src_port, _packet) do
+    ice_agent
+  end
+
   def handle_udp(ice_agent, socket, src_ip, src_port, packet) do
     turn_tr_id = {socket, {src_ip, src_port}}
     turn_tr = Map.get(ice_agent.gathering_transactions, turn_tr_id)
@@ -588,19 +632,58 @@ defmodule ExICE.Priv.ICEAgent do
         handle_turn_gathering_transaction_response(ice_agent, turn_tr_id, turn_tr, packet)
 
       from_turn?(ice_agent, src_ip, src_port) ->
-        handle_turn_message(ice_agent, socket, src_ip, src_port, packet)
+        local_cands = Map.values(ice_agent.local_cands)
+
+        case find_relay_cand_by_socket(local_cands, socket) do
+          nil ->
+            Logger.debug("""
+            Couldn't find relay candidate for:
+            socket: #{inspect(socket)}
+            src address: #{inspect({src_ip, src_port})}.
+            Ignoring incoming TURN message.
+            """)
+
+            ice_agent
+
+          relay_cand ->
+            handle_turn_message(ice_agent, relay_cand, src_ip, src_port, packet)
+        end
 
       ExSTUN.stun?(packet) ->
-        handle_stun_message(ice_agent, socket, src_ip, src_port, packet)
+        local_cands = Map.values(ice_agent.local_cands)
+
+        case find_host_cand(local_cands, socket) do
+          nil ->
+            Logger.debug("""
+            Couldn't find host candidate for #{inspect(src_ip)}:#{src_port}. \
+            Ignoring incoming STUN message.\
+            """)
+
+            ice_agent
+
+          host_cand ->
+            handle_stun_message(ice_agent, host_cand, src_ip, src_port, packet)
+        end
 
       true ->
-        local_cand = find_host_cand(Map.values(ice_agent.local_cands), socket)
-        remote_cand = find_remote_cand(Map.values(ice_agent.remote_cands), src_ip, src_port)
+        with remote_cands <- Map.values(ice_agent.remote_cands),
+             %_{} = local_cand <- find_host_cand(Map.values(ice_agent.local_cands), socket),
+             %_{} = remote_cand <- find_remote_cand(remote_cands, src_ip, src_port) do
+          %CandidatePair{} =
+            pair = Checklist.find_pair(ice_agent.checklist, local_cand.base.id, remote_cand.id)
 
-        %CandidatePair{} =
-          pair = Checklist.find_pair(ice_agent.checklist, local_cand.base.id, remote_cand.id)
+          handle_data_message(ice_agent, pair, packet)
+        else
+          _ ->
+            Logger.debug("""
+            Couldn't find host or remote candidate for:
+            socket: #{inspect(socket)}
+            src address: #{inspect({src_ip, src_port})}.
+            Ignoring incoming data message.
+            """)
 
-        handle_data_message(ice_agent, pair, packet)
+            ice_agent
+        end
     end
   end
 
@@ -900,9 +983,7 @@ defmodule ExICE.Priv.ICEAgent do
     end
   end
 
-  defp handle_turn_message(ice_agent, socket, src_ip, src_port, packet) do
-    %cand_mod{} = cand = find_relay_cand_by_socket(Map.values(ice_agent.local_cands), socket)
-
+  defp handle_turn_message(ice_agent, %cand_mod{} = cand, src_ip, src_port, packet) do
     case cand_mod.receive_data(cand, src_ip, src_port, packet) do
       {:ok, cand} ->
         put_in(ice_agent.local_cands[cand.base.id], cand)
@@ -935,23 +1016,10 @@ defmodule ExICE.Priv.ICEAgent do
     end
   end
 
-  defp handle_stun_message(ice_agent, socket, src_ip, src_port, packet) do
+  defp handle_stun_message(ice_agent, host_cand, src_ip, src_port, packet) do
     case ExSTUN.Message.decode(packet) do
       {:ok, msg} ->
-        local_cands = Map.values(ice_agent.local_cands)
-
-        case find_host_cand(local_cands, socket) do
-          nil ->
-            Logger.debug("""
-            Couldn't find host candidate for #{inspect(src_ip)}:#{src_port}. \
-            Ignoring incoming STUN message.\
-            """)
-
-            ice_agent
-
-          local_cand ->
-            do_handle_stun_message(ice_agent, local_cand, src_ip, src_port, msg)
-        end
+        do_handle_stun_message(ice_agent, host_cand, src_ip, src_port, msg)
 
       {:error, reason} ->
         Logger.warning("Couldn't decode stun message: #{inspect(reason)}")
@@ -1957,7 +2025,36 @@ defmodule ExICE.Priv.ICEAgent do
 
   @doc false
   @spec change_connection_state(t(), atom()) :: t()
+  def change_connection_state(ice_agent, :failed) do
+    Enum.each(ice_agent.sockets, fn socket ->
+      :ok = ice_agent.transport_module.close(socket)
+    end)
+
+    ice_agent = %{
+      ice_agent
+      | sockets: [],
+        gathering_transactions: %{},
+        selected_pair_id: nil,
+        conn_checks: %{},
+        checklist: %{},
+        local_cands: %{},
+        remote_cands: %{},
+        local_ufrag: nil,
+        local_pwd: nil,
+        remote_ufrag: nil,
+        remote_pwd: nil,
+        eoc: false,
+        nominating?: {false, nil}
+    }
+
+    do_change_connection_state(ice_agent, :failed)
+  end
+
   def change_connection_state(ice_agent, new_conn_state) do
+    do_change_connection_state(ice_agent, new_conn_state)
+  end
+
+  defp do_change_connection_state(ice_agent, new_conn_state) do
     Logger.debug("Connection state change: #{ice_agent.state} -> #{new_conn_state}")
     notify(ice_agent.on_connection_state_change, {:connection_state_change, new_conn_state})
     %__MODULE__{ice_agent | state: new_conn_state}

test/priv/ice_agent_test.exs

@@ -952,6 +952,59 @@ defmodule ExICE.Priv.ICEAgentTest do
     assert ice_agent.state == :failed
   end
 
+  @tag :debug
+  test "cleans up agent when moving to the failed state" do
+    remote_cand = ExICE.Candidate.new(:host, address: {192, 168, 0, 3}, port: 8445)
+
+    ice_agent =
+      ICEAgent.new(
+        controlling_process: self(),
+        role: :controlling,
+        transport_module: Transport.Mock,
+        if_discovery_module: IfDiscovery.Mock
+      )
+      |> ICEAgent.set_remote_credentials("remoteufrag", "remotepwd")
+      |> ICEAgent.gather_candidates()
+      |> ICEAgent.add_remote_candidate(remote_cand)
+
+    local_ufrag = ice_agent.local_ufrag
+
+    [socket] = ice_agent.sockets
+
+    # mark pair as failed
+    [pair] = Map.values(ice_agent.checklist)
+    ice_agent = put_in(ice_agent.checklist[pair.id], %{pair | state: :failed})
+
+    ice_agent = ICEAgent.end_of_candidates(ice_agent)
+
+    # agent should move to the failed state
+    assert ice_agent.state == :failed
+    assert ice_agent.sockets == []
+    assert ice_agent.local_cands == %{}
+    assert ice_agent.remote_cands == %{}
+
+    # assert that handle_udp ignores incoming data i.e. the state of ice_agent didn't change
+    new_ice_agent =
+      ICEAgent.handle_udp(ice_agent, socket, remote_cand.address, remote_cand.port, "some data")
+
+    assert ice_agent == new_ice_agent
+
+    # the same with incoming binding request
+    req =
+      binding_request(
+        ice_agent.role,
+        ice_agent.tiebreaker,
+        local_ufrag,
+        "remoteufrag",
+        "remotepwd"
+      )
+
+    new_ice_agent =
+      ICEAgent.handle_udp(ice_agent, socket, remote_cand.address, remote_cand.port, req)
+
+    assert ice_agent == new_ice_agent
+  end
+
   @stun_ip {192, 168, 0, 3}
   @stun_ip_str :inet.ntoa(@stun_ip)
   @stun_port 19_302
@@ -1439,6 +1492,28 @@ defmodule ExICE.Priv.ICEAgentTest do
     Message.new(%Type{class: :indication, method: :binding}) |> Message.encode()
   end
 
+  defp binding_request(role, tiebreaker, local_ufrag, remote_ufrag, remote_pwd) do
+    ice_attrs =
+      if role == :controlled do
+        [%ICEControlling{tiebreaker: tiebreaker + 1}, %UseCandidate{}]
+      else
+        [%ICEControlled{tiebreaker: tiebreaker - 1}]
+      end
+
+    attrs =
+      [
+        %Username{value: "#{remote_ufrag}:#{local_ufrag}"},
+        %Priority{priority: 1234}
+      ] ++ ice_attrs
+
+    request =
+      Message.new(%Type{class: :request, method: :binding}, attrs)
+      |> Message.with_integrity(remote_pwd)
+      |> Message.with_fingerprint()
+
+    Message.encode(request)
+  end
+
   defp binding_response(t_id, transport_module, socket, remote_pwd) do
     {:ok, {sock_ip, sock_port}} = transport_module.sockname(socket)