Improve rtsp decapsulator

lib/membrane/rtp/rtsp_decapsulator.ex

@@ -13,14 +13,7 @@ defmodule Membrane.RTP.RTSP.Decapsulator do
 
   alias Membrane.{Buffer, RemoteStream, RTP, RTSP}
 
-  def_options rtp_channel_id: [
-                spec: non_neg_integer(),
-                default: 0,
-                description: """
-                Channel identifier which encapsulated RTP packets will have.
-                """
-              ],
-              rtsp_session: [
+  def_options rtsp_session: [
                 spec: pid() | nil,
                 default: nil,
                 description: """
@@ -58,64 +51,71 @@ defmodule Membrane.RTP.RTSP.Decapsulator do
 
   @impl true
   def handle_buffer(:input, %Buffer{payload: payload, metadata: metadata}, _ctx, state) do
-    unprocessed_data =
-      if rtsp_response?(state.unprocessed_data, payload) do
-        if state.rtsp_session != nil do
-          {:ok, %RTSP.Response{status: 200}} =
-            RTSP.handle_response(state.rtsp_session, state.unprocessed_data)
-        end
-
-        <<>>
-      else
-        state.unprocessed_data
-      end
-
-    packets_binary = unprocessed_data <> payload
+    packets_binary = state.unprocessed_data <> payload
 
     {unprocessed_data, complete_packets_binaries} =
-      get_complete_packets(packets_binary, state.rtp_channel_id)
+      get_complete_packets(packets_binary, state.rtsp_session)
 
     packets_buffers =
       Enum.map(complete_packets_binaries, &%Buffer{payload: &1, metadata: metadata})
 
     {[buffer: {:output, packets_buffers}], %{state | unprocessed_data: unprocessed_data}}
   end
 
-  @spec rtsp_response?(binary(), binary()) :: boolean()
-  defp rtsp_response?(maybe_rtsp_response, new_payload) do
-    String.starts_with?(new_payload, "$") and String.starts_with?(maybe_rtsp_response, "RTSP")
-  end
-
-  @spec get_complete_packets(binary(), non_neg_integer()) ::
+  @spec get_complete_packets(binary(), pid() | nil, [binary()]) ::
           {unprocessed_data :: binary(), complete_packets :: [binary()]}
-  defp get_complete_packets(packets_binary, channel_id, complete_packets \\ [])
+  defp get_complete_packets(packets_binary, rtsp_session, complete_packets \\ [])
 
-  defp get_complete_packets(packets_binary, _channel_id, complete_packets)
+  defp get_complete_packets(packets_binary, _rtsp_session, complete_packets)
        when byte_size(packets_binary) <= 4 do
     {packets_binary, Enum.reverse(complete_packets)}
   end
 
   defp get_complete_packets(
-         <<"$", received_channel_id, payload_length::size(16), rest::binary>> = packets_binary,
-         channel_id,
+         <<"$", _channel_id, payload_length::size(16), rest::binary>> = packets_binary,
+         rtsp_session,
          complete_packets
        ) do
     case rest do
-      <<complete_packet_binary::binary-size(payload_length)-unit(8), rest::binary>> ->
-        complete_packets =
-          if channel_id != received_channel_id,
-            do: complete_packets,
-            else: [complete_packet_binary | complete_packets]
+      <<complete_packet_binary::binary-size(payload_length), rest::binary>> ->
+        complete_packets = [complete_packet_binary | complete_packets]
 
-        get_complete_packets(rest, channel_id, complete_packets)
+        get_complete_packets(rest, rtsp_session, complete_packets)
 
       _incomplete_packet_binary ->
         {packets_binary, Enum.reverse(complete_packets)}
     end
   end
 
-  defp get_complete_packets(rtsp_message, _channel_id, _complete_packets_binaries) do
-    # If the payload doesn't start with a "$" then it must be a RTSP message (or a part of it)
-    {rtsp_message, []}
+  defp get_complete_packets(
+         <<"RTSP", _rest::binary>> = rtsp_message_start,
+         rtsp_session,
+         complete_packets_binaries
+       ) do
+    case RTSP.Response.verify_content_length(rtsp_message_start) do
+      {:ok, _expected_length, _actual_length} ->
+        if rtsp_session != nil do
+          {:ok, %RTSP.Response{status: 200}} =
+            RTSP.handle_response(rtsp_session, rtsp_message_start)
+        end
+
+        {<<>>, complete_packets_binaries}
+
+      {:error, expected_length, actual_length} when actual_length > expected_length ->
+        rest_length = actual_length - expected_length
+        rtsp_message_length = byte_size(rtsp_message_start) - rest_length
+
+        <<rtsp_message::binary-size(rtsp_message_length)-unit(8), rest::binary>> =
+          rtsp_message_start
+
+        if rtsp_session != nil do
+          {:ok, %RTSP.Response{status: 200}} = RTSP.handle_response(rtsp_session, rtsp_message)
+        end
+
+        get_complete_packets(rest, rtsp_session, complete_packets_binaries)
+
+      {:error, expected_length, actual_length} when actual_length <= expected_length ->
+        {rtsp_message_start, Enum.reverse(complete_packets_binaries)}
+    end
   end
 end

mix.exs

@@ -40,7 +40,8 @@ defmodule Membrane.RTP.Plugin.MixProject do
       {:membrane_rtp_format, "~> 0.8.0"},
       {:membrane_funnel_plugin, "~> 0.9.0"},
       {:membrane_telemetry_metrics, "~> 0.1.0"},
-      {:membrane_rtsp, "~> 0.6.0"},
+      {:membrane_rtsp,
+       github: "membraneframework/membrane_rtsp", branch: "expand-length-verification"},
       {:ex_libsrtp, "~> 0.6.0 or ~> 0.7.0", optional: true},
       {:qex, "~> 0.5.1"},
       {:bunch, "~> 1.5"},

mix.lock

@@ -15,7 +15,7 @@
   "ex_doc": {:hex, :ex_doc, "0.31.1", "8a2355ac42b1cc7b2379da9e40243f2670143721dd50748bf6c3b1184dae2089", [:mix], [{:earmark_parser, "~> 1.4.39", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.1", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "3178c3a407c557d8343479e1ff117a96fd31bafe52a039079593fb0524ef61b0"},
   "ex_libsrtp": {:hex, :ex_libsrtp, "0.7.2", "211bd89c08026943ce71f3e2c0231795b99cee748808ed3ae7b97cd8d2450b6b", [:mix], [{:bunch, "~> 1.6", [hex: :bunch, repo: "hexpm", optional: false]}, {:bundlex, "~> 1.3", [hex: :bundlex, repo: "hexpm", optional: false]}, {:membrane_precompiled_dependency_provider, "~> 0.1.0", [hex: :membrane_precompiled_dependency_provider, repo: "hexpm", optional: false]}, {:unifex, "~> 1.1", [hex: :unifex, repo: "hexpm", optional: false]}], "hexpm", "2e20645d0d739a4ecdcf8d4810a0c198120c8a2f617f2b75b2e2e704d59f492a"},
   "ex_pcap": {:git, "https://github.com/membraneframework/expcap.git", "07c5bfa25280ea6a28d022d3a206ececf9b9913a", []},
-  "ex_sdp": {:hex, :ex_sdp, "0.14.1", "3c49b0a39c919713a3691a5c0637954680e3ad9aec6f87dc9e5a01978a32124a", [:mix], [{:bunch, "~> 1.3", [hex: :bunch, repo: "hexpm", optional: false]}, {:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}], "hexpm", "a4d6637b8eda59f218368a677caecffc054f6fc3228bedf17aeb6bfa8a58a6ed"},
+  "ex_sdp": {:hex, :ex_sdp, "0.15.0", "53815fb5b5e4fae0f3b26de90f372446bb8e0eed62a3cc20394d3c29519698be", [:mix], [{:bunch, "~> 1.3", [hex: :bunch, repo: "hexpm", optional: false]}, {:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}], "hexpm", "d3f23596b73e7057521ff0f0d55b1189c6320a2f04388aa3a80a0aa97ffb379f"},
   "file_system": {:hex, :file_system, "1.0.0", "b689cc7dcee665f774de94b5a832e578bd7963c8e637ef940cd44327db7de2cd", [:mix], [], "hexpm", "6752092d66aec5a10e662aefeed8ddb9531d79db0bc145bb8c40325ca1d8536d"},
   "finch": {:hex, :finch, "0.17.0", "17d06e1d44d891d20dbd437335eebe844e2426a0cd7e3a3e220b461127c73f70", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: false]}, {:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.3", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 0.2.6 or ~> 1.0", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "8d014a661bb6a437263d4b5abf0bcbd3cf0deb26b1e8596f2a271d22e48934c7"},
   "hackney": {:hex, :hackney, "1.20.1", "8d97aec62ddddd757d128bfd1df6c5861093419f8f7a4223823537bad5d064e2", [:rebar3], [{:certifi, "~> 2.12.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "~> 6.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "~> 1.0.0", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~> 1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.4.1", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~> 1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "fe9094e5f1a2a2c0a7d10918fee36bfec0ec2a979994cff8cfe8058cd9af38e3"},
@@ -40,7 +40,7 @@
   "membrane_rtp_format": {:hex, :membrane_rtp_format, "0.8.0", "828924bbd27efcf85b2015ae781e824c4a9928f0a7dc132abc66817b2c6edfc4", [:mix], [{:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}], "hexpm", "bc75d2a649dfaef6df563212fbb9f9f62eebc871393692f9dae8d289bd4f94bb"},
   "membrane_rtp_h264_plugin": {:hex, :membrane_rtp_h264_plugin, "0.19.0", "112bfedc14fb83bdb549ef1a03da23908feedeb165fd3e4512a549f1af532ae7", [:mix], [{:bunch, "~> 1.5", [hex: :bunch, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}, {:membrane_h264_format, "~> 0.6.0", [hex: :membrane_h264_format, repo: "hexpm", optional: false]}, {:membrane_rtp_format, "~> 0.8.0", [hex: :membrane_rtp_format, repo: "hexpm", optional: false]}], "hexpm", "76fd159e7406cadbef15124cba30eca3fffcf71a7420964f26e23d4cffd9b29d"},
   "membrane_rtp_mpegaudio_plugin": {:hex, :membrane_rtp_mpegaudio_plugin, "0.14.0", "eda543add9e035c8c0a99aaf7400f424922737efe934856b02b48e61679edba5", [:mix], [{:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}, {:membrane_mpegaudio_format, "~> 0.3.0", [hex: :membrane_mpegaudio_format, repo: "hexpm", optional: false]}, {:membrane_rtp_format, "~> 0.8.0", [hex: :membrane_rtp_format, repo: "hexpm", optional: false]}], "hexpm", "ec19247c7907771d38914aafbb7df74bdc2744c8b2ae29b34106b9bbcc5ede10"},
-  "membrane_rtsp": {:hex, :membrane_rtsp, "0.6.0", "aa4c26cf527711d36315950ca1b50a889183d3c8c53565a64ecd8a80e6289c6c", [:mix], [{:bunch, "~> 1.6", [hex: :bunch, repo: "hexpm", optional: false]}, {:ex_sdp, "~> 0.14.1", [hex: :ex_sdp, repo: "hexpm", optional: false]}, {:mockery, "~> 2.3", [hex: :mockery, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.4.0", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "66725d542f83bbcd1dd23fd28c44a6d7f49c859ac22d00876bd8d6de24744027"},
+  "membrane_rtsp": {:git, "https://github.com/membraneframework/membrane_rtsp.git", "fb7526292de77826fde814de8d81a8ee03441069", [branch: "expand-length-verification"]},
   "membrane_telemetry_metrics": {:hex, :membrane_telemetry_metrics, "0.1.0", "cb93d28356b436b0597736c3e4153738d82d2a14ff547f831df7e9051e54fc06", [:mix], [{:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 0.6.1", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}], "hexpm", "aba28dc8311f70ced95d984509be930fac55857d2d18bffcf768815e627be3f0"},
   "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"},
   "mime": {:hex, :mime, "2.0.5", "dc34c8efd439abe6ae0343edbb8556f4d63f178594894720607772a041b04b02", [:mix], [], "hexpm", "da0d64a365c45bc9935cc5c8a7fc5e49a0e0f9932a761c55d6c52b142780a05c"},