Handle DTLS close_notify alert

Author: mickel8

examples/whip_whep/lib/whip_whep/forwarder.ex

@@ -94,7 +94,7 @@ defmodule WhipWhep.Forwarder do
   @impl true
   def handle_info({:ex_webrtc, pc, {:connection_state_change, :failed}}, state) do
     Logger.info("Output peer connection (#{inspect(pc)}) state change: failed. Removing.")
-    :ok = PeerConnection.close(pc)
+    :ok = PeerConnection.stop(pc)
     pending_outputs = List.delete(state.pending_outputs, pc)
     outputs = Map.delete(state.outputs, pc)
     state = %{state | pending_outputs: pending_outputs, outputs: outputs}

lib/ex_webrtc/dtls_transport.ex

@@ -101,6 +101,11 @@ defmodule ExWebRTC.DTLSTransport do
     GenServer.cast(dtls_transport, {:set_packet_loss, packet_loss})
   end
 
+  @spec close(dtls_transport()) :: :ok
+  def close(dtls_transport) do
+    GenServer.call(dtls_transport, :close)
+  end
+
   @spec stop(dtls_transport()) :: :ok
   def stop(dtls_transport) do
     GenServer.stop(dtls_transport)
@@ -146,7 +151,7 @@ defmodule ExWebRTC.DTLSTransport do
     state = %{state | ice_connected: true}
 
     if state.mode == :active do
-      {packets, timeout} = ExDTLS.do_handshake(state.dtls)
+      {:ok, packets, timeout} = ExDTLS.do_handshake(state.dtls)
       Process.send_after(self(), :dtls_timeout, timeout)
       :ok = do_send(state, packets)
       state = update_dtls_state(state, :connecting)
@@ -158,7 +163,7 @@ defmodule ExWebRTC.DTLSTransport do
   end
 
   @impl true
-  def handle_call(:set_ice_connected, _from, state) do
+  def handle_call(:set_ice_connected, _from, %{dtls_state: :connecting} = state) do
     state = %{state | ice_connected: true}
 
     if state.buffered_local_packets do
@@ -171,6 +176,17 @@ defmodule ExWebRTC.DTLSTransport do
     end
   end
 
+  @impl true
+  def handle_call(:set_ice_connected, _from, state) do
+    Logger.debug("""
+    Setting ice connected in unexpected DTLS state: #{state.dtls_state}. \
+    DTLS handshake won't be performed.\
+    """)
+
+    state = %{state | ice_connected: true}
+    {:reply, :ok, state}
+  end
+
   @impl true
   def handle_call(:get_certs_info, _from, state) do
     local_cert_info = %{
@@ -239,6 +255,19 @@ defmodule ExWebRTC.DTLSTransport do
     {:reply, {:error, :already_started}, state}
   end
 
+  @impl true
+  def handle_call(:close, _from, %{state: :closed} = state) do
+    {:reply, :ok, state}
+  end
+
+  @impl true
+  def handle_call(:close, _from, state) do
+    {:ok, packets} = ExDTLS.disconnect(state.dtls)
+    :ok = do_send(state, packets)
+    state = update_dtls_state(state, :closed, notify: false)
+    {:reply, :ok, state}
+  end
+
   @impl true
   def handle_cast({:send_rtp, data}, %{dtls_state: :connected, ice_connected: true} = state) do
     case ExLibSRTP.protect(state.out_srtp, data) do
@@ -327,6 +356,14 @@ defmodule ExWebRTC.DTLSTransport do
       {:ok, state} ->
         {:noreply, state}
 
+      {:error, :peer_closed_for_writing} ->
+        # See W3C WebRTC sec. 5.5.1
+        # peer_closed_for_writing is returned when the remote side
+        # sends close_notify alert
+        ExDTLS.disconnect(state.dtls)
+        state = update_dtls_state(state, :closed)
+        {:noreply, state}
+
       {:error, _reason} ->
         # See W3C WebRTC sec. 5.5.
         state = update_dtls_state(state, :failed)
@@ -345,6 +382,12 @@ defmodule ExWebRTC.DTLSTransport do
     Logger.debug("Stopping DTLSTransport with reason: #{inspect(reason)}")
   end
 
+  defp handle_ice_data({:data, _data}, %{dtls_state: dtls_state} = state)
+       when dtls_state in [:failed, :closed] do
+    Logger.debug("Received DTLS packets in state #{dtls_state}. Ignoring.")
+    {:ok, state}
+  end
+
   defp handle_ice_data({:data, data}, %{dtls: nil} = state) do
     # received DTLS data from remote peer before receiving an
     # SDP answer and initializing the DTLS Transport, will buffer the data
@@ -470,11 +513,16 @@ defmodule ExWebRTC.DTLSTransport do
     :ok
   end
 
-  defp update_dtls_state(%{dtls_state: dtls_state} = state, dtls_state), do: state
+  defp update_dtls_state(state, dtls_state, otps \\ [])
+  defp update_dtls_state(%{dtls_state: dtls_state} = state, dtls_state, _opts), do: state
 
-  defp update_dtls_state(state, new_dtls_state) do
+  defp update_dtls_state(state, new_dtls_state, opts) do
     Logger.debug("Changing DTLS state: #{state.dtls_state} -> #{new_dtls_state}")
-    notify(state.owner, {:state_change, new_dtls_state})
+
+    if opts[:notify] != false do
+      notify(state.owner, {:state_change, new_dtls_state})
+    end
+
     %{state | dtls_state: new_dtls_state}
   end
 

lib/ex_webrtc/ice_transport.ex

@@ -3,7 +3,7 @@ defmodule ExWebRTC.ICETransport do
 
   # module implementing this behaviour
   @type t() :: module()
-  @type state() :: :checking | :connected | :completed | :failed
+  @type state() :: :checking | :connected | :completed | :failed | :closed
 
   @callback start_link(Keyword.t()) :: {:ok, pid()}
   @callback on_data(pid(), pid()) :: :ok
@@ -19,6 +19,7 @@ defmodule ExWebRTC.ICETransport do
   @callback set_role(pid(), ExICE.ICEAgent.role()) :: :ok
   @callback set_remote_credentials(pid(), ufrag :: binary(), pwd :: binary()) :: :ok
   @callback get_stats(pid()) :: map()
+  @callback close(pid()) :: :ok
   @callback stop(pid()) :: :ok
 end
 
@@ -58,5 +59,7 @@ defmodule ExWebRTC.DefaultICETransport do
   @impl true
   defdelegate get_stats(pid), to: ICEAgent
   @impl true
+  def close(_pid), do: :ok
+  @impl true
   defdelegate stop(pid), to: ICEAgent
 end

lib/ex_webrtc/peer_connection.ex

@@ -63,7 +63,7 @@ defmodule ExWebRTC.PeerConnection do
 
   For the exact meaning, refer to the [RTCDtlsTransport: state property](https://developer.mozilla.org/en-US/docs/Web/API/RTCDtlsTransport/state)
   """
-  @type dtls_transport_state() :: :new | :connecting | :connected | :failed
+  @type dtls_transport_state() :: :new | :connecting | :connected | :closed | :failed
 
   @typedoc """
   Possible signaling states.
@@ -105,8 +105,8 @@ defmodule ExWebRTC.PeerConnection do
            | {:track_muted, MediaStreamTrack.id()}
            | {:track_ended, MediaStreamTrack.id()}
            | {:data, DataChannel.ref(), binary()}
-           | {:rtp, MediaStreamTrack.id(), String.t() | nil, ExRTP.Packet.t()}}
-          | {:rtcp, [{MediaStreamTrack.id() | nil, ExRTCP.Packet.packet()}]}
+           | {:rtp, MediaStreamTrack.id(), String.t() | nil, ExRTP.Packet.t()}
+           | {:rtcp, [{MediaStreamTrack.id() | nil, ExRTCP.Packet.packet()}]}}
 
   #### NON-MDN-API ####
 
@@ -472,7 +472,7 @@ defmodule ExWebRTC.PeerConnection do
           peer_connection(),
           RTPTransceiver.kind() | MediaStreamTrack.t(),
           direction: RTPTransceiver.direction()
-        ) :: {:ok, RTPTransceiver.t()}
+        ) :: {:ok, RTPTransceiver.t()} | {:error, term()}
   def add_transceiver(peer_connection, kind_or_track, options \\ []) do
     GenServer.call(peer_connection, {:add_transceiver, kind_or_track, options})
   end
@@ -507,7 +507,8 @@ defmodule ExWebRTC.PeerConnection do
 
   For more information, refer to the [RTCPeerConnection: addTrack() method](https://developer.mozilla.org/en-US/docs/Web/API/RTCPeerConnection/addTrack).
   """
-  @spec add_track(peer_connection(), MediaStreamTrack.t()) :: {:ok, RTPSender.t()}
+  @spec add_track(peer_connection(), MediaStreamTrack.t()) ::
+          {:ok, RTPSender.t()} | {:error, term()}
   def add_track(peer_connection, track) do
     GenServer.call(peer_connection, {:add_track, track})
   end
@@ -574,11 +575,20 @@ defmodule ExWebRTC.PeerConnection do
   @doc """
   Closes the PeerConnection.
 
-  This function kills the `peer_connection` process.
+  This function doest not kill the `peer_connection` process.
+  If you want to stop the `peer_connection` process, see `stop/1`.
   For more information, refer to the [RTCPeerConnection: close() method](https://developer.mozilla.org/en-US/docs/Web/API/RTCPeerConnection/close).
   """
   @spec close(peer_connection()) :: :ok
   def close(peer_connection) do
+    GenServer.call(peer_connection, :close)
+  end
+
+  @doc """
+  Closes and stops PeerConnection process.
+  """
+  @spec stop(peer_connection()) :: :ok
+  def stop(peer_connection) do
     GenServer.stop(peer_connection)
   end
 
@@ -1067,6 +1077,14 @@ defmodule ExWebRTC.PeerConnection do
 
   if Code.ensure_loaded?(ExSCTP) do
     @impl true
+    def handle_call(
+          {:create_data_channel, _label, _opts},
+          _from,
+          %{connection_state: :closed} = state
+        ) do
+      {:reply, {:error, :closed}, state}
+    end
+
     def handle_call({:create_data_channel, label, opts}, _from, state) do
       ordered = Keyword.get(opts, :ordered, true)
       lifetime = Keyword.get(opts, :max_packet_life_time)
@@ -1075,7 +1093,7 @@ defmodule ExWebRTC.PeerConnection do
 
       with true <- byte_size(label) < 65_535,
            true <- lifetime == nil or max_rtx == nil do
-        {events, channel, sctp_transport} =
+        {:ok, events, channel, sctp_transport} =
           SCTPTransport.add_channel(
             state.sctp_transport,
             label,
@@ -1271,9 +1289,17 @@ defmodule ExWebRTC.PeerConnection do
     {:reply, stats, state}
   end
 
+  @impl true
+  def handle_call(:close, _from, state) do
+    Logger.debug("Closing peer connection")
+    # don't emit state change events
+    state = do_close(state, notify: false)
+    {:reply, :ok, state}
+  end
+
   @impl true
   def handle_cast({:send_rtp, track_id, packet, opts}, %{conn_state: conn_state} = state)
-      when conn_state != :failed do
+      when conn_state not in [:failed, :closed] do
     rtx? = Keyword.get(opts, :rtx?, false)
 
     # TODO: iterating over transceivers is not optimal
@@ -1331,7 +1357,7 @@ defmodule ExWebRTC.PeerConnection do
 
   @impl true
   def handle_cast({:send_pli, track_id, rid}, %{conn_state: conn_state} = state)
-      when conn_state != :failed do
+      when conn_state not in [:failed, :closed] do
     state.transceivers
     |> Enum.with_index()
     |> Enum.find(fn {tr, _idx} -> tr.receiver.track.id == track_id end)
@@ -1365,7 +1391,7 @@ defmodule ExWebRTC.PeerConnection do
   if Code.ensure_loaded?(ExSCTP) do
     @impl true
     def handle_cast({:send_data, channel_ref, data_type, data}, %{conn_state: conn_state} = state)
-        when conn_state != :failed do
+        when conn_state not in [:failed, :closed] do
       {events, sctp_transport} =
         SCTPTransport.send(state.sctp_transport, channel_ref, data_type, data)
 
@@ -1432,12 +1458,17 @@ defmodule ExWebRTC.PeerConnection do
 
     next_conn_state = next_conn_state(state.ice_state, new_dtls_state)
 
-    state =
-      %{state | dtls_state: new_dtls_state}
-      |> update_conn_state(next_conn_state)
-      |> maybe_connect_sctp()
+    if next_conn_state == :closed and state.conn_state != :closed do
+      state = do_close(state)
+      {:noreply, state}
+    else
+      state =
+        %{state | dtls_state: new_dtls_state}
+        |> update_conn_state(next_conn_state)
+        |> maybe_connect_sctp()
 
-    {:noreply, state}
+      {:noreply, state}
+    end
   end
 
   @impl true
@@ -1955,15 +1986,24 @@ defmodule ExWebRTC.PeerConnection do
   defp next_signaling_state(:have_remote_pranswer, :remote, :answer), do: {:ok, :stable}
   defp next_signaling_state(:have_remote_pranswer, _, _), do: {:error, :invalid_state}
 
-  defp update_signaling_state(%{signaling_state: signaling_state} = state, signaling_state),
-    do: state
+  defp update_signaling_state(state, signaling_state, opts \\ [])
+
+  defp update_signaling_state(
+         %{signaling_state: signaling_state} = state,
+         signaling_state,
+         _opts
+       ),
+       do: state
 
-  defp update_signaling_state(state, new_signaling_state) do
+  defp update_signaling_state(state, new_signaling_state, opts) do
     Logger.debug(
       "Changing PeerConnection signaling state state: #{state.signaling_state} -> #{new_signaling_state}"
     )
 
-    notify(state.owner, {:signaling_state_change, new_signaling_state})
+    if opts[:notify] != false do
+      notify(state.owner, {:signaling_state_change, new_signaling_state})
+    end
+
     %{state | signaling_state: new_signaling_state}
   end
 
@@ -2192,12 +2232,15 @@ defmodule ExWebRTC.PeerConnection do
   defp get_last_answer(%{current_remote_desc: {:answer, desc}}), do: desc
 
   # TODO support :disconnected state - our ICE doesn't provide disconnected state for now
-  # TODO support :closed state
   # the order of these clauses is important
   defp next_conn_state(ice_state, dtls_state)
 
-  defp next_conn_state(ice_state, dtls_state) when ice_state == :failed or dtls_state == :failed,
-    do: :failed
+  # Give closed precedence over failed.
+  # Failed connection can be restarted.
+  # Closed connection can't be reused.
+  defp next_conn_state(:closed, _dtls_state), do: :closed
+
+  defp next_conn_state(_ice_state, :closed), do: :closed
 
   defp next_conn_state(:failed, _dtls_state), do: :failed
 
@@ -2212,11 +2255,16 @@ defmodule ExWebRTC.PeerConnection do
   defp next_conn_state(ice_state, :connected) when ice_state in [:connected, :completed],
     do: :connected
 
-  defp update_conn_state(%{conn_state: conn_state} = state, conn_state), do: state
+  defp update_conn_state(state, conn_state, opts \\ [])
+  defp update_conn_state(%{conn_state: conn_state} = state, conn_state, _opts), do: state
 
-  defp update_conn_state(state, new_conn_state) do
+  defp update_conn_state(state, new_conn_state, opts) do
     Logger.debug("Changing PeerConnection state: #{state.conn_state} -> #{new_conn_state}")
-    notify(state.owner, {:connection_state_change, new_conn_state})
+
+    if opts[:notify] != false do
+      notify(state.owner, {:connection_state_change, new_conn_state})
+    end
+
     %{state | conn_state: new_conn_state}
   end
 
@@ -2413,6 +2461,23 @@ defmodule ExWebRTC.PeerConnection do
     %SessionDescription{type: type, sdp: to_string(sdp)}
   end
 
+  defp do_close(state, opts \\ []) do
+    transceivers = Enum.map(state.transceivers, &RTPTransceiver.stop(&1))
+    sctp_transport = SCTPTransport.close_abruptly(state.sctp_transport)
+    :ok = DTLSTransport.close(state.dtls_transport)
+    :ok = state.ice_transport.close(state.ice_pid)
+
+    %{
+      state
+      | ice_state: :closed,
+        dtls_state: :closed,
+        transceivers: transceivers,
+        sctp_transport: sctp_transport
+    }
+    |> update_signaling_state(:closed, opts)
+    |> update_conn_state(:closed, opts)
+  end
+
   defp generate_ssrcs(state) do
     generate_ssrcs(state.transceivers, Map.keys(state.demuxer.ssrc_to_mid))
   end

lib/ex_webrtc/rtp_transceiver.ex

@@ -469,7 +469,7 @@ defmodule ExWebRTC.RTPTransceiver do
 
   @doc false
   @spec stop(transceiver(), (-> term())) :: transceiver()
-  def stop(transceiver, on_track_ended) do
+  def stop(transceiver, on_track_ended \\ fn -> :ok end) do
     transceiver =
       if transceiver.stopping,
         do: transceiver,