defmodule ExLibnice do @moduledoc """ Module that wraps functions from [libnice](https://libnice.freedesktop.org/libnice/index.html). For more information about each function please refer to [libnice] documentation. """ use GenServer use Bunch require Logger require Unifex.CNode defmodule State do @moduledoc false use Bunch.Access @type t :: %__MODULE__{ parent: pid, impl: NIF | CNode, cnode: Unifex.CNode.t(), native_state: reference(), stream_components: %{stream_id: integer(), n_components: integer()}, mdns_queries: %{ query: String.t(), candidates: [{sdp :: String.t(), stream_id :: integer(), component_id :: integer()}] } } @enforce_keys [:parent, :impl] defstruct @enforce_keys ++ [ cnode: nil, native_state: nil, stream_components: %{}, mdns_queries: %{} ] end @typedoc """ Fully qualified domain name e.g. "my.domain.com". """ @type fqdn() :: String.t() @type stun_server() :: %{server_addr: :inet.ip_address() | fqdn(), server_port: 0..65_535} @typedoc """ Type describing TURN server configuration """ @type relay_info :: %{ server_addr: :inet.ip_address() | fqdn(), server_port: 0..65_535, username: String.t(), password: String.t(), relay_type: :udp | :tcp | :tls } @typedoc """ Type describing ExLibnice configuration. It's a keyword list containing the following keys: * impl - implementation to use. Possible values are NIF and CNode. You can also choose `impl` via config.exs by ```elixir config :ex_libnice, impl: :NIF ``` * parent - pid of calling process * stun_servers - list of stun servers in form of ip:port * controlling_mode - refer to RFC 8445 section 4 - Controlling and Controlled Agent * min_port..max_port - the port range to use. Pass 0..0 if you not willing to set it. Passed port range will be set for each newly added stream. At this moment it is not possible to set port range per stream. """ @type opts_t :: [ impl: NIF | CNode, parent: pid(), stun_servers: [stun_server()], controlling_mode: boolean(), port_range: 0..65_535 ] @doc """ Spawns new process responsible for interacting with `libnice`. """ @spec start_link(opts :: opts_t) :: {:ok, pid()} def start_link(opts) do GenServer.start_link(__MODULE__, opts) end @doc """ Adds a new stream with `n_components` components. `name` is optional but not passing it will influence on SDP string format, see `generate_local_sdp/1`. """ @spec add_stream(pid :: pid(), n_components :: integer(), name :: String.t()) :: {:ok, integer()} | {:error, :failed_to_add_stream | :invalid_stream_or_duplicate_name | :failed_to_attach_recv} def add_stream(pid, n_components, name \\ "") do GenServer.call(pid, {:add_stream, n_components, name}) end @doc """ Removes stream with id `stream_id`. """ @spec remove_stream(pid :: pid(), stream_id :: integer()) :: :ok def remove_stream(pid, stream_id) do GenServer.cast(pid, {:remove_stream, stream_id}) end @doc """ Sets TURN server. Can be called multiple times for the same component to add many TURN servers. """ @spec set_relay_info( pid :: pid(), stream_id :: integer(), component_id :: integer() | [integer()] | :all, relay_info :: relay_info() | [relay_info()] ) :: :ok | {:error, :bad_stream_id | :bad_relay_type | :failed_to_resolve_addr | :failed_to_set_turn} def set_relay_info(pid, stream_id, component_id, relay_info) do GenServer.call(pid, {:set_relay_info, stream_id, component_id, relay_info}) end @doc """ Forget all TURN servers for given component. """ @spec forget_relays(pid :: pid(), stream_id :: integer(), component_id :: integer()) :: :ok | {:error, :component_not_found} def forget_relays(pid, stream_id, component_id) do GenServer.call(pid, {:forget_relays, stream_id, component_id}) end @doc """ Generates a SDP string containing the local candidates and credentials for all streams and components. Notice that: - local candidates will be present in the result SDP only if `gather_candidates/1` has been invoked previously - if a stream is created without a name the 'm' line will contain '-' mark but SDP in this form will not be parsable by `parse_remote_sdp/2` """ @spec generate_local_sdp(pid :: pid()) :: {:ok, sdp :: String.t()} def generate_local_sdp(pid) do GenServer.call(pid, :generate_local_sdp) end @doc """ Parses a remote SDP string setting credentials and remote candidates for proper streams and components. It is important that `m` line does not contain `-` mark but the name of the stream. """ @spec parse_remote_sdp(pid :: pid(), sdp :: String.t()) :: {:ok, added_cand_num :: integer()} | {:error, :failed_to_parse_sdp} def parse_remote_sdp(pid, sdp) do GenServer.call(pid, {:parse_remote_sdp, sdp}) end @doc """ Returns local credentials for stream with id `stream_id`. """ @spec get_local_credentials(pid :: pid(), stream_id :: integer()) :: {:ok, credentials :: String.t()} | {:error, :failed_to_get_credentials} def get_local_credentials(pid, stream_id) do GenServer.call(pid, {:get_local_credentials, stream_id}) end @doc """ Sets remote credentials for stream with `stream_id`. Credentials have to be passed in form of `ufrag pwd`. """ @spec set_remote_credentials(pid :: pid(), credentials :: String.t(), stream_id :: integer()) :: :ok | {:error, :failed_to_set_credentials} def set_remote_credentials(pid, credentials, stream_id) do GenServer.call(pid, {:set_remote_credentials, credentials, stream_id}) end @doc """ Starts gathering candidates process for stream with id `stream_id`. May cause the parent process receive following messages: - `{:new_candidate_full, candidate}` - new local candidate - `{:candidate_gathering_done, stream_id}` - gathering candidates for stream with `stream_id` has been done """ @spec gather_candidates(pid :: pid(), stream_id :: integer()) :: :ok | {:error, :invalid_stream_or_allocation} def gather_candidates(pid, stream_id) do GenServer.call(pid, {:gather_candidates, stream_id}) end @doc """ Indicates that all remote candidates for stream with id `stream_id` have been passed. After receiving this message components can change their state to `FAILED` if all their connectivity checks have failed. Not sending this message will cause components stay in `CONNECTING` state. (In fact there is a bug and components can change their state to `FAILED` even without sending this message. Please refer to [#120](https://gitlab.freedesktop.org/libnice/libnice/-/issues/120.) """ @spec peer_candidate_gathering_done(pid :: pid(), stream_id :: integer()) :: :ok | {:error, :stream_not_found} def peer_candidate_gathering_done(pid, stream_id) do GenServer.call(pid, {:peer_candidate_gathering_done, stream_id}) end @doc """ Sets remote candidate for component with id `component_id` in stream with id `stream_id`. Candidate has to be passed as SDP string. May cause the parent process receive following messages: - `{:new_remote_candidate_full, candidate}` - new remote (prflx) candidate - `{:new_selected_pair, stream_id, component_id, lfoundation, rfoundation}` - new selected pair - `{:component_state_failed, stream_id, component_id}` - component with id `component_id` in stream with id `stream_id` has changed state to FAILED - `{:component_state_ready, stream_id, component_id}` - component with id `component_id` in stream with id `stream_id` has changed state to READY i.e. it is ready to receive and send data """ @spec set_remote_candidate( pid :: pid(), candidate :: String.t(), stream_id :: integer(), component_id :: integer() ) :: :ok | {:error, :failed_to_parse_sdp_string | :failed_to_set} def set_remote_candidate(pid, candidate, stream_id, component_id) do GenServer.call(pid, {:set_remote_candidate, candidate, stream_id, component_id}) end @doc """ Restarts all streams. """ @spec restart(pid :: pid()) :: :ok | {:error, :failed_to_restart} def restart(pid) do GenServer.call(pid, :restart) end @doc """ Restarts stream with id `stream_id`. """ @spec restart_stream(pid :: pid(), stream_id :: integer()) :: :ok | {:error, :failed_to_restart} def restart_stream(pid, stream_id) do GenServer.call(pid, {:restart_stream, stream_id}) end @doc """ Sends payload on component with id `component_id` in stream with id `stream_id`. Payload has to be in a binary format. """ @spec send_payload( pid :: pid(), stream_id :: integer(), component_id :: integer(), payload :: binary() ) :: :ok | {:error, :failed_to_send} def send_payload(pid, stream_id, component_id, payload) do GenServer.call(pid, {:send_payload, stream_id, component_id, payload}) end @doc """ Stops and cleans up `ExLibnice` instance. """ @spec stop(pid :: pid(), reason :: term(), timeout :: timeout()) :: :ok def stop(pid, reason \\ :normal, timeout \\ :infinity) do GenServer.stop(pid, reason, timeout) end # Server API @impl true def init(opts) do min_port..max_port = opts[:port_range] {:ok, stun_servers} = lookup_stun_servers(opts[:stun_servers]) impl = Application.get_env(:ex_libnice, :impl) || opts[:impl] || CNode state = %State{parent: opts[:parent], impl: impl} {:ok, state} = call(impl, :init, [stun_servers, opts[:controlling_mode], min_port, max_port], state) {:ok, state} end @impl true def handle_call({:add_stream, n_components}, from, state) do handle_call({:add_stream, n_components, ""}, from, state) end @impl true def handle_call({:add_stream, n_components, name}, _from, state) do case call(state.impl, :add_stream, [n_components, name], state) do {{:ok, stream_id}, state} -> Logger.debug("New stream_id: #{stream_id}") {:reply, {:ok, stream_id}, put_in(state.stream_components[stream_id], n_components)} {{:error, cause}, state} -> Logger.warn(""" Couldn't add stream with #{n_components} components and name "#{inspect(name)}": \ #{inspect(cause)} """) {:reply, {:error, cause}, state} end end @impl true def handle_call({:set_relay_info, stream_id, component_id, relay_info}, _from, state) do {ret, state} = Bunch.listify(relay_info) |> Bunch.Enum.try_reduce(state, fn single_relay_info, state -> do_set_relay_info(state, stream_id, component_id, single_relay_info) end) {:reply, ret, state} end @impl true def handle_call({:forget_relays, stream_id, component_id}, _from, state) do case call(state.impl, :forget_relays, [stream_id, component_id], state) do {:ok, state} -> {:reply, :ok, state} {{:error, cause}, state} -> Logger.warn(""" Couldn't forget TURN servers for component: #{inspect(component_id)} in stream: \ #{inspect(stream_id)}, reason: #{inspect(cause)} """) {:reply, {:error, cause}, state} end end @impl true def handle_call(:generate_local_sdp, _from, state) do {{:ok, local_sdp}, state} = call(state.impl, :generate_local_sdp, [], state) Logger.debug("local sdp: #{inspect(local_sdp)}") {:reply, {:ok, local_sdp}, state} end @impl true def handle_call({:parse_remote_sdp, remote_sdp}, _from, state) do case call(state.impl, :parse_remote_sdp, [remote_sdp], state) do {{:ok, added_cand_num}, state} -> Logger.debug("parse_remote_sdp: ok; added #{added_cand_num} candidates") {:reply, {:ok, added_cand_num}, state} {{:error, cause}, state} -> Logger.warn("Couldn't parse remote sdp #{inspect(remote_sdp)}") {:reply, {:error, cause}, state} end end @impl true def handle_call({:get_local_credentials, stream_id}, _from, state) do case call(state.impl, :get_local_credentials, [stream_id], state) do {{:ok, credentials}, state} -> Logger.debug("local credentials: #{credentials}") {:reply, {:ok, credentials}, state} {{:error, cause}, state} -> Logger.error("get_local_credentials: #{inspect(cause)}") {{:error, cause}, state} end end @impl true def handle_call( {:set_remote_credentials, credentials, stream_id}, _from, state ) do case call(state.impl, :set_remote_credentials, [credentials, stream_id], state) do {:ok, state} -> Logger.debug("set_remote_credentials: ok") {:reply, :ok, state} {{:error, cause}, state} -> Logger.error("set_remote_credentials: #{inspect(cause)}") {:reply, {:error, cause}, state} end end @impl true def handle_call({:gather_candidates, stream_id} = msg, _from, state) do case call(state.impl, :gather_candidates, [stream_id], state) do {:ok, state} -> Logger.debug("#{inspect(msg)}") {:reply, :ok, state} {{:error, cause}, state} -> Logger.error("gather_candidates: #{inspect(msg)}") {:reply, {:error, cause}, state} end end @impl true def handle_call( {:peer_candidate_gathering_done, stream_id}, _ctx, state ) do case call(state.impl, :peer_candidate_gathering_done, [stream_id], state) do {:ok, state} -> Logger.debug("peer_candidate_gathering_done: ok") {:reply, :ok, state} {{:error, cause}, state} -> Logger.warn("peer_candidate_gathering_done: #{inspect(cause)}") {:reply, {:error, cause}, state} end end @impl true def handle_call( {:set_remote_candidate, "a=", _stream_id, _component_id}, _from, state ) do Logger.debug("Empty candidate \"a=\". Should we do something with it?") {:reply, :ok, state} end @impl true def handle_call({:set_remote_candidate, candidate, stream_id, component_id}, _from, state) do candidate_sp = String.split(candidate, " ", parts: 6) withl candidate_check: 6 <- length(candidate_sp), do: address = Enum.at(candidate_sp, 4), mdns_check: true <- String.ends_with?(address, ".local") do if Application.get_env(:ex_libnice, :mdns, true) do state = update_in(state, [:mdns_queries, address], fn # if we haven't query this address yet nil -> ExLibnice.Mdns.query(address) [{candidate, stream_id, component_id}] candidates -> [{candidate, stream_id, component_id} | candidates] end) {:reply, :ok, state} else Logger.debug("Got mdns address but mdns client is turned off. Ignoring.") {:reply, :ok, state} end else candidate_check: _ -> {:reply, {:error, :failed_to_parse_sdp_string}, state} mdns_check: _ -> do_set_remote_candidate(candidate, stream_id, component_id, state) end end @impl true def handle_call(:restart, _from, state) do case call(state.impl, :restart, [], state) do {:ok, state} -> Logger.debug("ICE restarted") {:reply, :ok, state} {{:error, cause}, state} -> Logger.warn("Couldn't restart ICE") {:reply, {:error, cause}, state} end end @impl true def handle_call({:restart_stream, stream_id}, _from, state) do case call(state.impl, :restart_stream, [stream_id], state) do {:ok, state} -> Logger.debug("Stream #{inspect(stream_id)} restarted") {:reply, :ok, state} {{:error, cause}, state} -> Logger.warn("Couldn't restart stream #{inspect(stream_id)}") {:reply, {:error, cause}, state} end end @impl true def handle_call({:send_payload, stream_id, component_id, payload}, _from, state) do case call(state.impl, :send_payload, [stream_id, component_id, payload], state) do {:ok, state} -> {:reply, :ok, state} {{:error, cause}, state} -> Logger.warn("Couldn't send payload: #{inspect(cause)}") {:reply, {:error, cause}, state} end end @impl true def handle_cast({:remove_stream, stream_id}, state) do {:ok, state} = call(state.impl, :remove_stream, [stream_id], state) Logger.debug("remove_stream #{stream_id}: ok") {_n_components, state} = pop_in(state.stream_components[stream_id]) {:noreply, state} end @impl true def handle_info({:new_candidate_full, _cand} = msg, %State{parent: parent} = state) do Logger.debug("#{inspect(msg)}") send(parent, msg) {:noreply, state} end @impl true def handle_info({:new_remote_candidate_full, _cand} = msg, %State{parent: parent} = state) do Logger.debug("#{inspect(msg)}") send(parent, msg) {:noreply, state} end @impl true def handle_info({:candidate_gathering_done, _stream_id} = msg, %State{parent: parent} = state) do Logger.debug("#{inspect(msg)}") send(parent, msg) {:noreply, state} end @impl true def handle_info( {:new_selected_pair, _stream_id, _component_id, _lfoundation, _rfoundation} = msg, %State{parent: parent} = state ) do Logger.debug("#{inspect(msg)}") send(parent, msg) {:noreply, state} end @impl true def handle_info( {:component_state_ready, _stream_id, _component_id} = msg, %State{parent: parent} = state ) do Logger.debug("#{inspect(msg)}") send(parent, msg) {:noreply, state} end @impl true def handle_info( {:component_state_failed, _stream_id, _component_id} = msg, %State{parent: parent} = state ) do Logger.warn("#{inspect(msg)}") send(parent, msg) {:noreply, state} end @impl true def handle_info( {:ice_payload, _stream_id, _component_id, _payload} = msg, %State{parent: parent} = state ) do send(parent, msg) {:noreply, state} end @impl true def handle_info({:mdns_response, address, ip}, state) do {candidates, state} = pop_in(state, [:mdns_queries, address]) for {candidate, stream_id, component_id} <- candidates do candidate_parts = String.split(candidate, " ", parts: 6) |> List.replace_at(4, :inet.ntoa(ip)) candidate = Enum.join(candidate_parts, " ") do_set_remote_candidate(candidate, stream_id, component_id, state) end {:noreply, state} end @impl true def handle_info(msg, state) do Logger.warn("Unknown message #{inspect(msg)}") {:noreply, state} end @impl true def terminate(_reason, %State{native_state: nil, cnode: cnode}) do Logger.debug("Terminating ExLibnice instance #{inspect(self())}") Unifex.CNode.stop(cnode) end @impl true def terminate(_reason, _state) do Logger.debug("Terminating ExLibnice instance #{inspect(self())}") :ok end defp do_set_relay_info(state, stream_id, n_components, relay_info) when is_list(n_components), do: Bunch.Enum.try_reduce(n_components, state, fn n_component, state -> do_set_relay_info(state, stream_id, n_component, relay_info) end) defp do_set_relay_info(state, stream_id, :all, relay_info) do case Map.get(state.stream_components, stream_id) do nil -> Logger.warn("Couldn't set TURN servers. No stream with id #{inspect(stream_id)}") {{:error, :bad_stream_id}, state} n_components -> Bunch.Enum.try_reduce(1..n_components, state, fn n_component, state -> do_set_relay_info(state, stream_id, n_component, relay_info) end) end end defp do_set_relay_info( state, stream_id, component_id, %{server_addr: server_addr, server_port: server_port, relay_type: relay_type} ) when relay_type not in [:udp, :tcp, :tls] do Logger.warn(""" Couldn't set TURN server #{inspect(server_addr)} #{inspect(server_port)} \ #{inspect(relay_type)} for component: #{inspect(component_id)} in stream: \ #{inspect(stream_id)}, cause: bad_relay_type """) {{:error, :bad_relay_type}, state} end defp do_set_relay_info(state, stream_id, component_id, %{ server_addr: server_addr, server_port: server_port, username: username, password: password, relay_type: relay_type }) do with {:ok, server_ip} <- lookup_addr(server_addr), {:ok, state} <- call( state.impl, :set_relay_info, [ stream_id, component_id, :inet.ntoa(server_ip) |> to_string(), server_port, username, password, Atom.to_string(relay_type) ], state ) do {:ok, state} else {:error, :failed_to_lookup_addr} = err -> Logger.warn(""" Couldn't lookup TURN server address #{inspect(server_addr)} """) {err, state} {{:error, cause}, _state} = ret -> Logger.warn(""" Couldn't set TURN server #{inspect(server_addr)} #{inspect(server_port)} \ #{inspect(relay_type)} for component: #{inspect(component_id)} in stream: \ #{inspect(stream_id)}, cause: #{inspect(cause)} """) ret end end defp do_set_remote_candidate(candidate, stream_id, component_id, state) do case call(state.impl, :set_remote_candidate, [candidate, stream_id, component_id], state) do {:ok, state} -> Logger.debug("Set remote candidate: #{inspect(candidate)}") {:reply, :ok, state} {{:error, cause}, state} -> Logger.warn("Couldn't set remote candidate: #{inspect(cause)}") {:reply, {:error, cause}, state} end end defp call(NIF, :init = func, args, state) do {ret, native_state} = apply(ExLibnice.Native, func, args) {ret, %{state | native_state: native_state}} end defp call(NIF, func, args, state) do {ret, native_state} = apply(ExLibnice.Native, func, [state.native_state | args]) {ret, %{state | native_state: native_state}} end defp call(CNode, func, args, %{cnode: nil} = state) do {:ok, cnode} = Unifex.CNode.start_link(:native) call(CNode, func, args, %{state | cnode: cnode}) end defp call(CNode, func, args, state) do ret = apply(Unifex.CNode, :call, [state.cnode, func, args]) {ret, state} end defp lookup_stun_servers(stun_servers) do Bunch.Enum.try_map(stun_servers, fn %{server_addr: addr, server_port: port} -> case lookup_addr(addr) do {:ok, ip} -> {:ok, "#{:inet.ntoa(ip)}:#{port}"} {:error, cause} -> {:error, cause, addr} end end) end defp lookup_addr({_a, _b, _c, _d} = addr), do: {:ok, addr} defp lookup_addr({_a, _b, _c, _d, _e, _f, _g, _h} = addr), do: {:ok, addr} defp lookup_addr(addr) when is_binary(addr) do case :inet_res.lookup(to_charlist(addr), :in, :a) do [] -> {:error, :failed_to_lookup_address} [h | _t] -> {:ok, h} end end end