sdpdemux: Add support for RFC4570 SDP source filters

Parse source-filter attributes and configure udpsrc accordingly

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/3485>

subprojects/gst-plugins-bad/gst/sdp/gstsdpdemux.c

@@ -421,6 +421,9 @@ gst_sdp_demux_stream_free (GstSDPDemux * demux, GstSDPStream * stream)
     }
     stream->srcpad = NULL;
   }
+
+  g_free (stream->src_list);
+  g_free (stream->src_incl_list);
   g_free (stream);
 }
 
@@ -455,6 +458,149 @@ out:
   return ret;
 }
 
+/* RTC 4570 Session Description Protocol (SDP) Source Filters
+ * syntax:
+ * a=source-filter: <filter-mode> <filter-spec>
+ *
+ * where
+ * <filter-mode>: "incl" or "excl"
+ *
+ * <filter-spec>:
+ * <nettype> <address-types> <dest-address> <src-list>
+ *
+ */
+static gboolean
+gst_sdp_demux_parse_source_filter (GstSDPDemux * self,
+    const gchar * source_filter, const gchar * dst_addr, GString * source_list,
+    GString * source_incl_list)
+{
+  const gchar *str;
+  guint remaining;
+  gchar *del;
+  gsize size;
+  guint min_size;
+  gboolean is_incl;
+  gchar *dst;
+
+  if (!source_filter || !dst_addr)
+    return FALSE;
+
+  str = source_filter;
+  remaining = strlen (str);
+  min_size = strlen ("incl IN IP4 * *");
+  if (remaining < min_size)
+    return FALSE;
+
+#define LSTRIP(s) G_STMT_START { \
+  while (g_ascii_isspace (*(s))) { \
+    (s)++; \
+    remaining--; \
+  } \
+  if (*(s) == '\0') \
+    return FALSE; \
+} G_STMT_END
+
+#define SKIP_N_LSTRIP(s, n) G_STMT_START { \
+  if (remaining < n) \
+    return FALSE; \
+  (s) += n; \
+  if (*(s) == '\0') \
+    return FALSE; \
+  remaining -= n; \
+  LSTRIP(s); \
+} G_STMT_END
+
+  LSTRIP (str);
+  if (remaining < min_size)
+    return FALSE;
+
+  if (g_str_has_prefix (str, "incl ")) {
+    is_incl = TRUE;
+  } else if (g_str_has_prefix (str, "excl ")) {
+    is_incl = FALSE;
+  } else {
+    GST_WARNING_OBJECT (self, "Unexpected filter type");
+    return FALSE;
+  }
+
+  SKIP_N_LSTRIP (str, 4);
+  /* XXX: <nettype>, internet only for now */
+  if (!g_str_has_prefix (str, "IN "))
+    return FALSE;
+
+  SKIP_N_LSTRIP (str, 3);
+  /* Should care the address type here? */
+  if (g_str_has_prefix (str, "* ")) {
+    /* dest and src are both FQDN */
+    SKIP_N_LSTRIP (str, 2);
+  } else if (g_str_has_prefix (str, "IP4 ")) {
+    SKIP_N_LSTRIP (str, 4);
+  } else if (g_str_has_prefix (str, "IP6 ")) {
+    SKIP_N_LSTRIP (str, 4);
+  } else {
+    return FALSE;
+  }
+
+  del = strchr (str, ' ');
+  if (!del) {
+    GST_WARNING_OBJECT (self, "Unexpected dest-address format");
+    return FALSE;
+  }
+
+  size = del - str;
+  dst = g_strndup (str, size);
+  if (g_strcmp0 (dst, dst_addr) != 0 && g_strcmp0 (dst, "*") != 0) {
+    g_free (dst);
+    return FALSE;
+  }
+  g_free (dst);
+
+  SKIP_N_LSTRIP (str, size);
+
+  do {
+    del = strchr (str, ' ');
+    if (del) {
+      size = del - str;
+      if (is_incl) {
+        g_string_append_c (source_list, '+');
+        g_string_append_len (source_list, str, size);
+
+        g_string_append_c (source_incl_list, '+');
+        g_string_append_len (source_incl_list, str, size);
+      } else {
+        g_string_append_c (source_list, '-');
+        g_string_append_len (source_list, str, size);
+      }
+
+      str += size;
+      while (g_ascii_isspace (*str)) {
+        str++;
+      }
+
+      /* this was the last source but with trailing space */
+      if (*str == '\0')
+        return TRUE;
+    } else {
+      if (is_incl) {
+        g_string_append_c (source_list, '+');
+        g_string_append (source_list, str);
+
+        g_string_append_c (source_incl_list, '+');
+        g_string_append (source_incl_list, str);
+      } else {
+        g_string_append_c (source_list, '-');
+        g_string_append (source_list, str);
+      }
+
+      return TRUE;
+    }
+  } while (TRUE);
+
+#undef LSTRIP
+#undef SKIP_N
+  return TRUE;
+}
+
 static GstSDPStream *
 gst_sdp_demux_create_stream (GstSDPDemux * demux, GstSDPMessage * sdp, gint idx)
 {
@@ -531,6 +677,48 @@ gst_sdp_demux_create_stream (GstSDPDemux * demux, GstSDPMessage * sdp, gint idx)
   stream->destination = conn->address;
   stream->ttl = conn->ttl;
   stream->multicast = is_multicast_address (stream->destination);
+  if (stream->multicast) {
+    GString *source_list = g_string_new (NULL);
+    GString *source_incl_list = g_string_new (NULL);
+    guint i;
+    gboolean source_filter_in_media = FALSE;
+
+    for (i = 0; i < media->attributes->len; i++) {
+      GstSDPAttribute *attr = &g_array_index (media->attributes,
+          GstSDPAttribute, i);
+
+      if (g_strcmp0 (attr->key, "source-filter") == 0) {
+        source_filter_in_media = TRUE;
+        gst_sdp_demux_parse_source_filter (demux, attr->value,
+            stream->destination, source_list, source_incl_list);
+      }
+    }
+
+    /* Try session level source filter if media level filter is unspecified */
+    if (source_list->len == 0 && !source_filter_in_media) {
+      for (i = 0; i < sdp->attributes->len; i++) {
+        GstSDPAttribute *attr = &g_array_index (sdp->attributes,
+            GstSDPAttribute, i);
+
+        if (g_strcmp0 (attr->key, "source-filter") == 0) {
+          gst_sdp_demux_parse_source_filter (demux, attr->value,
+              stream->destination, source_list, source_incl_list);
+        }
+      }
+    }
+
+    if (source_list->len > 0) {
+      stream->src_list = g_string_free (source_list, FALSE);
+      stream->src_incl_list = g_string_free (source_incl_list, FALSE);
+
+      GST_DEBUG_OBJECT (demux,
+          "Have source-filter: \"%s\", positive-only: \"%s\"",
+          stream->src_list, GST_STR_NULL (stream->src_incl_list));
+    } else {
+      g_string_free (source_list, TRUE);
+      g_string_free (source_incl_list, TRUE);
+    }
+  }
 
   stream->rtp_port = gst_sdp_media_get_port (media);
 
@@ -866,7 +1054,13 @@ gst_sdp_demux_stream_configure_udp (GstSDPDemux * demux, GstSDPStream * stream)
     GST_DEBUG_OBJECT (demux, "receiving RTP from %s:%d", destination,
         stream->rtp_port);
 
-    uri = g_strdup_printf ("udp://%s:%d", destination, stream->rtp_port);
+    if (stream->src_list) {
+      uri = g_strdup_printf ("udp://%s:%d?multicast-source=%s",
+          destination, stream->rtp_port, stream->src_list);
+    } else {
+      uri = g_strdup_printf ("udp://%s:%d", destination, stream->rtp_port);
+    }
+
     stream->udpsrc[0] =
         gst_element_make_from_uri (GST_URI_SRC, uri, NULL, NULL);
     g_free (uri);
@@ -909,7 +1103,13 @@ gst_sdp_demux_stream_configure_udp (GstSDPDemux * demux, GstSDPStream * stream)
           || demux->rtcp_mode == GST_SDP_DEMUX_RTCP_MODE_RECVONLY)) {
     GST_DEBUG_OBJECT (demux, "receiving RTCP from %s:%d", destination,
         stream->rtcp_port);
-    uri = g_strdup_printf ("udp://%s:%d", destination, stream->rtcp_port);
+    /* rfc4570 3.2.1. Source-Specific Multicast Example */
+    if (stream->src_incl_list) {
+      uri = g_strdup_printf ("udp://%s:%d?multicast-source=%s",
+          destination, stream->rtcp_port, stream->src_incl_list);
+    } else {
+      uri = g_strdup_printf ("udp://%s:%d", destination, stream->rtcp_port);
+    }
     stream->udpsrc[1] =
         gst_element_make_from_uri (GST_URI_SRC, uri, NULL, NULL);
     g_free (uri);

subprojects/gst-plugins-bad/gst/sdp/gstsdpdemux.h

@@ -72,6 +72,10 @@ struct _GstSDPStream {
   guint         ttl;
   gboolean      multicast;
 
+  /* source-filter */
+  gchar        *src_list;
+  gchar        *src_incl_list;
+
   /* our udp sink back to the server */
   GstElement   *udpsink;
   GstPad       *rtcppad;

subprojects/gst-plugins-bad/tests/check/elements/sdpdemux.c

@@ -0,0 +1,186 @@
+/*
+ * GStreamer
+ * Copyright (C) 2023 Seungha Yang <seungha@centricular.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gst/gst.h>
+#include "../../../gst/sdp/gstsdpdemux.h"
+#include "../../../gst/sdp/gstsdpdemux.c"
+#undef GST_CAT_DEFAULT
+
+#include <gst/check/gstcheck.h>
+
+static void
+test_source_filter (const gchar * str, guint size, const gchar * result)
+{
+  GstSDPMessage sdp = { 0 };
+  GstSDPDemux *demux = g_object_new (GST_TYPE_SDP_DEMUX, NULL);
+  GstSDPStream *stream;
+
+  gst_sdp_message_init (&sdp);
+  fail_if (gst_sdp_message_parse_buffer ((const guint8 *) str, size,
+          &sdp) != GST_SDP_OK);
+
+  stream = gst_sdp_demux_create_stream (demux, &sdp, 0);
+  fail_if (stream == NULL);
+
+  if (result) {
+    fail_unless_equals_string (stream->src_list, result);
+  } else {
+    fail_if (stream->src_list);
+  }
+
+  gst_sdp_demux_cleanup (demux);
+  gst_object_unref (demux);
+  gst_sdp_message_uninit (&sdp);
+}
+
+GST_START_TEST (test_parse_source_filter_incl)
+{
+  static const gchar sdp[] =
+      "v=0\r\n"
+      "o=- 18 0 IN IP4 127.0.0.1\r\n"
+      "s=TestSdp\r\n"
+      "t=0 0\r\n"
+      "m=audio 5004 RTP/AVP 98\r\n"
+      "c=IN IP4 224.0.0.0\r\n"
+      "a=recvonly\r\n"
+      "a=source-filter: incl IN IP4 224.0.0.0 127.0.0.1\r\n"
+      "a=rtpmap:98 L24/48000/2\r\n" "a=framecount:48\r\n" "a=recvonly\r\n";
+
+  test_source_filter (sdp, sizeof (sdp), "+127.0.0.1");
+}
+
+GST_END_TEST;
+
+GST_START_TEST (test_parse_source_filter_incl_multi_list)
+{
+  static const gchar sdp[] =
+      "v=0\r\n"
+      "o=- 18 0 IN IP4 127.0.0.1\r\n"
+      "s=TestSdp\r\n"
+      "t=0 0\r\n"
+      "m=audio 5004 RTP/AVP 98\r\n"
+      "c=IN IP4 224.0.0.0\r\n"
+      "a=recvonly\r\n"
+      "a=source-filter: incl IN IP4 224.0.0.0 127.0.0.1 127.0.0.2\r\n"
+      "a=rtpmap:98 L24/48000/2\r\n" "a=framecount:48\r\n" "a=recvonly\r\n";
+
+  test_source_filter (sdp, sizeof (sdp), "+127.0.0.1+127.0.0.2");
+}
+
+GST_END_TEST;
+
+GST_START_TEST (test_parse_source_filter_excl)
+{
+  static const gchar sdp[] =
+      "v=0\r\n"
+      "o=- 18 0 IN IP4 127.0.0.1\r\n"
+      "s=TestSdp\r\n"
+      "t=0 0\r\n"
+      "m=audio 5004 RTP/AVP 98\r\n"
+      "c=IN IP4 224.0.0.0\r\n"
+      "a=recvonly\r\n"
+      "a=source-filter: excl IN IP4 224.0.0.0 127.0.0.2\r\n"
+      "a=rtpmap:98 L24/48000/2\r\n" "a=framecount:48\r\n" "a=recvonly\r\n";
+
+  test_source_filter (sdp, sizeof (sdp), "-127.0.0.2");
+}
+
+GST_END_TEST;
+
+GST_START_TEST (test_parse_source_filter_incl_excl)
+{
+  static const gchar sdp[] =
+      "v=0\r\n"
+      "o=- 18 0 IN IP4 127.0.0.1\r\n"
+      "s=TestSdp\r\n"
+      "t=0 0\r\n"
+      "m=audio 5004 RTP/AVP 98\r\n"
+      "c=IN IP4 224.0.0.0\r\n"
+      "a=recvonly\r\n"
+      "a=source-filter: incl IN IP4 224.0.0.0 127.0.0.1\r\n"
+      "a=source-filter: excl IN IP4 224.0.0.0 127.0.0.2\r\n"
+      "a=rtpmap:98 L24/48000/2\r\n" "a=framecount:48\r\n" "a=recvonly\r\n";
+
+  test_source_filter (sdp, sizeof (sdp), "+127.0.0.1-127.0.0.2");
+}
+
+GST_END_TEST;
+
+GST_START_TEST (test_parse_source_filter_with_trailing_space)
+{
+  static const gchar sdp[] =
+      "v=0\r\n"
+      "o=- 18 0 IN IP4 127.0.0.1\r\n"
+      "s=TestSdp\r\n"
+      "t=0 0\r\n"
+      "m=audio 5004 RTP/AVP 98\r\n"
+      "c=IN IP4 224.0.0.0\r\n"
+      "a=recvonly\r\n"
+      "a=source-filter: incl  IN   IP4  224.0.0.0   127.0.0.1    \r\n"
+      "a=rtpmap:98 L24/48000/2\r\n" "a=framecount:48\r\n" "a=recvonly\r\n";
+
+  test_source_filter (sdp, sizeof (sdp), "+127.0.0.1");
+}
+
+GST_END_TEST;
+
+GST_START_TEST (test_parse_source_filter_missing_list)
+{
+  static const gchar sdp[] =
+      "v=0\r\n"
+      "o=- 18 0 IN IP4 127.0.0.1\r\n"
+      "s=TestSdp\r\n"
+      "t=0 0\r\n"
+      "m=audio 5004 RTP/AVP 98\r\n"
+      "c=IN IP4 224.0.0.0\r\n"
+      "a=recvonly\r\n"
+      "a=source-filter: incl IN IP4 224.0.0.0 \r\n"
+      "a=rtpmap:98 L24/48000/2\r\n" "a=framecount:48\r\n" "a=recvonly\r\n";
+
+  test_source_filter (sdp, sizeof (sdp), NULL);
+}
+
+GST_END_TEST;
+
+static Suite *
+sdpdemux_suite (void)
+{
+  Suite *s;
+  TCase *tc_chain;
+
+  s = suite_create ("sdpdemux");
+  tc_chain = tcase_create ("general");
+
+  suite_add_tcase (s, tc_chain);
+  tcase_add_test (tc_chain, test_parse_source_filter_incl);
+  tcase_add_test (tc_chain, test_parse_source_filter_incl_multi_list);
+  tcase_add_test (tc_chain, test_parse_source_filter_excl);
+  tcase_add_test (tc_chain, test_parse_source_filter_incl_excl);
+  tcase_add_test (tc_chain, test_parse_source_filter_with_trailing_space);
+  tcase_add_test (tc_chain, test_parse_source_filter_missing_list);
+
+  return s;
+}
+
+GST_CHECK_MAIN (sdpdemux);

subprojects/gst-plugins-bad/tests/check/meson.build

@@ -68,6 +68,7 @@ base_tests = [
   [['elements/rtponviftimestamp.c'], get_option('onvif').disabled()],
   [['elements/rtpsrc.c'], get_option('rtp').disabled()],
   [['elements/rtpsink.c'], get_option('rtp').disabled()],
+  [['elements/sdpdemux.c'], get_option('sdp').disabled(), [gstsdp_dep]],
   [['elements/srtp.c'], not srtp_dep.found(), [srtp_dep]],
   [['elements/switchbin.c'], get_option('switchbin').disabled()],
   [['elements/videoframe-audiolevel.c'], get_option('videoframe_audiolevel').disabled()],