defmodule Membrane.RTC.Engine.Endpoint.SIP.Call do @moduledoc false require Membrane.Logger alias Membrane.Logger alias Membrane.RTC.Engine.Endpoint.SIP alias Membrane.RTC.Engine.Endpoint.SIP.Call.{Auth, Headers, Settings} alias Membrane.RTC.Engine.Endpoint.SIP.{CallRegistry, SippetCore} @type id :: String.t() defmodule State do @moduledoc false use Bunch.Access @type t :: %__MODULE__{ endpoint: pid(), rtp_port: 1..65_535, sip_port: 1..65_535, registrar_credentials: SIP.RegistrarCredentials.t(), external_ip: String.t(), register_interval_ms: non_neg_integer(), phone_number: String.t() | nil, call_id: SIP.Call.id(), callee: Sippet.URI.t() | nil, to: Sippet.Message.name_uri_params() | nil, target: {:udp, String.t(), non_neg_integer()} | nil, route: [Sippet.Message.name_uri_params()] | nil, headers_base: Headers.t(), cseq: non_neg_integer(), last_message: Sippet.Message.t() | nil, # Pending requests: # %{{cseq, method} => time when request was made # or when last provisional response to the request was received} pending_requests: %{{non_neg_integer(), atom()} => integer()}, invite_dialog_complete?: boolean() } @enforce_keys [ :endpoint, :rtp_port, :sip_port, :registrar_credentials, :external_ip, :register_interval_ms, :phone_number, :call_id, :callee, :to, :target, :route, :headers_base, :cseq, :last_message, :pending_requests, :invite_dialog_complete? ] defstruct @enforce_keys end @type state :: State.t() @callback start_link(Settings.t()) :: {id(), pid()} @callback after_init(state :: state()) :: state() @callback handle_request( method :: atom(), request :: Sippet.Message.request(), state :: state() ) :: state() @callback handle_response( method :: atom(), status_code :: pos_integer(), response :: Sippet.Message.response(), state :: state() ) :: state() defmacro __using__(_opts) do quote location: :keep do use GenServer require Membrane.Logger alias Membrane.Logger alias Membrane.RTC.Engine.Endpoint.SIP @behaviour SIP.Call @impl SIP.Call def start_link(settings) do call_id = Sippet.Message.create_call_id() Logger.debug("SIP Client: Starting call process with id #{inspect(call_id)}") {:ok, pid} = GenServer.start_link(__MODULE__, {call_id, settings}, name: SIP.Call.registry_id(call_id) ) {call_id, pid} end @impl SIP.Call def after_init(state) do state end @impl SIP.Call def handle_request(_method, request, state) do Logger.warning("SIP Client: handle_request not implemented for #{inspect(request)}") state end @impl SIP.Call def handle_response(_method, status_code, response, state) do SIP.Call.handle_generic_response(status_code, response, state) end defoverridable after_init: 1, handle_request: 3, handle_response: 4 @impl GenServer def init({call_id, settings}) do state = SIP.Call.init_state(call_id, settings) {:ok, __MODULE__.after_init(state)} end defguardp is_request_pending(state, cseq) when is_map_key(state.pending_requests, cseq) @impl GenServer def handle_cast({:response, %{headers: %{cseq: cseq}}}, state) when not is_request_pending(state, cseq) do Logger.warning( "SIP Client: Received response with CSeq #{inspect(cseq)}, for which there is no pending request. Ignoring." ) {:noreply, state} end @impl GenServer def handle_cast({:response, %{headers: %{cseq: {_cseq, method}}} = response}, state) do Logger.debug("Received response in call: #{inspect(response)}") state = SIP.Call.process_response(response, state) status_code = response.start_line.status_code state = __MODULE__.handle_response(method, status_code, response, state) {:noreply, state} end @impl GenServer def handle_cast({:request, %{headers: %{cseq: {_cseq, method}}} = request}, state) do Logger.debug("Received request in call: #{inspect(request)}") {:noreply, __MODULE__.handle_request(method, request, state)} end @impl GenServer def handle_info({:timeout, cseq}, state) when is_request_pending(state, cseq) do if SIP.Call.timeout?(cseq, state) do raise "SIP Client: Timeout. Received no response for request with CSeq #{inspect(cseq)}" end {:noreply, state} end @impl GenServer def handle_info({:timeout, _cseq}, state) do {:noreply, state} end end end @timeout_ms 32_000 ## MANAGEMENT API @spec stop(id()) :: :ok def stop(call_id) do GenServer.stop(registry_id(call_id)) end @spec exists?(id()) :: boolean() def exists?(call_id) do Registry.lookup(CallRegistry, call_id) != [] end ## INCOMING API @spec handle_request(id(), term()) :: :ok def handle_request(call_id, request) do GenServer.cast(registry_id(call_id), {:request, request}) end @spec handle_response(id(), term()) :: :ok def handle_response(call_id, response) do GenServer.cast(registry_id(call_id), {:response, response}) end ## PRIVATE API (for modules implementing the behaviour) @spec registry_id(id()) :: {:via, module(), term()} def registry_id(call_id) do {:via, Registry, {CallRegistry, call_id}} end @spec init_state(id(), Settings.t()) :: state() def init_state(call_id, settings) do from_address = %Sippet.URI{ scheme: "sip", userinfo: settings.registrar_credentials.username, host: settings.external_ip, port: settings.sip_port } callee = if is_nil(settings.phone_number), do: nil, else: %{settings.registrar_credentials.uri | userinfo: settings.phone_number} settings |> Map.from_struct() |> Map.merge(%{ call_id: call_id, callee: callee, to: nil, target: nil, route: nil, headers_base: Headers.create_headers_base(from_address), cseq: 0, last_message: nil, pending_requests: %{}, invite_dialog_complete?: false }) |> then(&struct!(State, &1)) end @spec build_headers(atom(), state(), map(), String.t()) :: map() def build_headers( method, state, override_headers \\ %{}, branch \\ Sippet.Message.create_branch() ) do # XXX: Consider optimising the digest auth process # (right now, we do the exchange `request, 401, request with digest` every time) headers = state.headers_base |> Map.merge(%{ to: state.to || {"", state.callee, %{}}, call_id: state.call_id, cseq: {state.cseq + 1, method}, content_length: 0 }) |> Map.merge(override_headers) |> update_in([:via], fn via -> [Tuple.append(via, %{"branch" => branch})] end) if is_nil(state.route) do headers else Map.put(headers, :route, state.route) end end @spec make_request(Sippet.Message.request(), state(), boolean()) :: state() | no_return() def make_request(message, state, use_target? \\ true) do message = if use_target? and state.target != nil, do: Map.put(message, :target, state.target), else: message with :ok <- SippetCore.send_message(message) do cseq = message.headers.cseq Process.send_after(self(), {:timeout, cseq}, @timeout_ms) pending_requests = Map.put(state.pending_requests, cseq, System.monotonic_time(:millisecond)) %{state | cseq: elem(cseq, 0), last_message: message, pending_requests: pending_requests} else error -> Logger.debug("Send failed with message: #{inspect(message)}") raise "SIP Client: Unable to send message: #{inspect(error)}" end end # Generic processing of _all_ responses @spec process_response(Sippet.Message.response(), state()) :: state() def process_response(response, state) do state |> update_routing(response) |> update_pending_requests(response) end @spec timeout?(non_neg_integer(), state()) :: boolean() | no_return() def timeout?(cseq, state) do System.monotonic_time(:millisecond) >= Map.fetch!(state.pending_requests, cseq) + @timeout_ms end @spec handle_generic_response(pos_integer(), Sippet.Message.response(), state()) :: state() | no_return() def handle_generic_response(status_code, response, state) do case status_code do success when success in [200, 204] -> state unauthorized when unauthorized in [401, 407] -> handle_unauthorized(response, state) redirect when redirect in 300..399 -> raise """ SIP Client: Received redirection response with code #{status_code}. Redirections need to be handled by the module implementing the `Call` behaviour """ _other_failure -> raise """ SIP Client: Received unhandled failure response with code #{status_code} (#{inspect(response.start_line.reason_phrase)}): #{inspect(response)} """ end end defp handle_unauthorized(response, state) do headers = state.last_message.headers authorization_headers = headers[:authorization] || headers[:proxy_authorization] || [] cond do not Enum.empty?(authorization_headers) -> raise "SIP Client: Unable to authorize using digest auth (incorrect credentials?)" Sippet.Message.has_header?(response, :www_authenticate) or Sippet.Message.has_header?(response, :proxy_authenticate) -> request = Auth.apply_digest(state.last_message, response, state.registrar_credentials) make_request(request, state) true -> raise "SIP Client: Unable to authorize using digest auth (no `www-authenticate` or `proxy-authenticate` header present)" end end defp update_pending_requests(state, response) do cseq = response.headers.cseq pending_requests = if response.start_line.status_code in 100..199 do Process.send_after(self(), {:timeout, cseq}, @timeout_ms) Map.put(state.pending_requests, cseq, System.monotonic_time(:millisecond)) else Map.delete(state.pending_requests, cseq) end %{state | pending_requests: pending_requests} end # According to RFC 3261 section 12.2.1.1 # https://datatracker.ietf.org/doc/html/rfc3261#section-12.2.1.1 defp update_routing(state, response) when is_map_key(response.headers, :record_route) do %Sippet.Message{ headers: %{ to: to, record_route: record_route, contact: [{_name, contact_uri, _params} | _] = contact } } = response route = Enum.reverse(record_route) [{_name, first_hop_uri, _params} | _] = route loose_routing? = loose_routing?(first_hop_uri) {state, route} = if loose_routing? do Logger.debug("SIP Client: using loose routing") {%{state | callee: contact_uri, target: {:udp, first_hop_uri.host, first_hop_uri.port}}, route} else Logger.debug("SIP Client: using strict routing") callee = first_hop_uri |> Map.put(:parameters, nil) {%{state | callee: callee}, Enum.drop(route, 1) ++ contact} end %{state | route: route, to: to} end defp update_routing(state, response) do state = %{state | to: response.headers.to} if is_map_key(response.headers, :contact) do [{_name, contact_uri, _params} | _] = response.headers.contact %{state | callee: contact_uri} else state end end defp loose_routing?(first_hop_uri) do params = Map.fetch!(first_hop_uri, :parameters) || "" params |> String.split(";") |> Enum.member?("lr") end end