srt: add authentication to srtsink and srtsrc elements

ext/srt/gstsrtobject.c

@@ -54,6 +54,8 @@ enum
   PROP_STATS,
   PROP_WAIT_FOR_CONNECTION,
   PROP_STREAMID,
+  PROP_AUTHENTICATION,
+
   PROP_LAST
 };
 
@@ -352,6 +354,9 @@ gst_srt_object_set_property_helper (GstSRTObject * srtobject,
     case PROP_STREAMID:
       gst_structure_set_value (srtobject->parameters, "streamid", value);
       break;
+    case PROP_AUTHENTICATION:
+      srtobject->authentication = g_value_get_boolean (value);
+      break;
     default:
       goto err;
   }
@@ -436,6 +441,9 @@ gst_srt_object_get_property_helper (GstSRTObject * srtobject,
           gst_structure_get_string (srtobject->parameters, "streamid"));
       break;
     }
+    case PROP_AUTHENTICATION:
+      g_value_set_boolean (value, srtobject->authentication);
+      break;
     default:
       goto err;
   }
@@ -583,6 +591,19 @@ gst_srt_object_install_properties_helper (GObjectClass * gobject_class)
           "Stream ID for the SRT access control", "",
           G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY |
           G_PARAM_STATIC_STRINGS));
+
+  /**
+   * GstSRTSink:authentication:
+   *
+   * Boolean to block streaming until a client connects.  If TRUE,
+   * `srtsink' will stream only when a client is connected.
+   */
+  g_object_class_install_property (gobject_class, PROP_AUTHENTICATION,
+      g_param_spec_boolean ("authentication",
+          "Enable authentication",
+          "Enable authentication",
+          FALSE,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 }
 
 static void
@@ -829,6 +850,84 @@ thread_func (gpointer data)
   }
 }
 
+static void
+parse_stream_id (const gchar * stream_id, gchar ** username, gchar ** resource)
+{
+  const gchar STREAM_ID_PREFIX[] = "#!::";
+  gchar **keys;
+  gchar **it;
+
+  if (!g_str_has_prefix (stream_id, STREAM_ID_PREFIX)) {
+    return;
+  }
+  if (username) {
+    *username = NULL;
+  }
+  if (resource) {
+    *resource = NULL;
+  }
+  keys = g_strsplit (stream_id + sizeof (STREAM_ID_PREFIX) - 1, ",", -1);
+  for (it = keys; *it; ++it) {
+    gchar **keyval;
+    keyval = g_strsplit (*it, "=", 2);
+    if (keyval && keyval[0] && keyval[1]) {
+      if (g_str_equal (keyval[0], "u") && username) {
+        g_clear_pointer (username, g_free);
+        *username = g_strdup (keyval[1]);
+      } else if (g_str_equal (keyval[0], "r") && resource) {
+        g_clear_pointer (resource, g_free);
+        *resource = g_strdup (keyval[1]);
+      }
+    }
+    g_strfreev (keyval);
+  }
+  g_strfreev (keys);
+}
+
+static GSocketAddress *
+peeraddr_to_g_socket_address (const struct sockaddr *peeraddr) {
+  gsize peeraddr_len;
+
+  switch (peeraddr->sa_family) {
+    case AF_INET:
+      peeraddr_len = sizeof (struct sockaddr_in);
+      break;
+    case AF_INET6:
+      peeraddr_len = sizeof (struct sockaddr_in6);
+      break;
+    default:
+      g_warning ("Unsupported address family %d", peeraddr->sa_family);
+      return NULL;
+  }
+  return g_socket_address_new_from_native ((gpointer) peeraddr, peeraddr_len);
+}
+
+static gint
+srt_listen_callback_func(GstSRTObject * self, SRTSOCKET sock, int hs_version,
+    const struct sockaddr* peeraddr, const char* stream_id) {
+  g_autoptr (GSocketAddress) addr = peeraddr_to_g_socket_address (peeraddr);
+  gchar *username = NULL;
+  g_autofree gchar *resource = NULL;
+
+  if (self->authentication) {
+    gboolean authenticated = FALSE;
+
+    parse_stream_id (stream_id, &username, &resource);
+    /* notifying caller-connecting */
+    g_signal_emit_by_name (self->element, "caller-connecting", addr,
+      username, &authenticated);
+
+    if (!authenticated)
+      goto reject;
+  }
+
+  return 0;
+reject:
+  /* notifying caller-rejected */
+  g_signal_emit_by_name (self->element, "caller-rejected");
+  return -1;
+}
+
 static gboolean
 gst_srt_object_wait_connect (GstSRTObject * srtobject,
     GCancellable * cancellable, GSocketFamily sa_family, gpointer sa,
@@ -910,14 +1009,18 @@ gst_srt_object_wait_connect (GstSRTObject * srtobject,
 
   srtobject->listener_sock = sock;
 
-  srtobject->thread =
-      g_thread_try_new ("GstSRTObjectListener", thread_func, srtobject, error);
+  /* Register the SRT listen callback */
+  if (!srt_listen_callback (srtobject->listener_sock,
+      (srt_listen_callback_fn *) srt_listen_callback_func, srtobject)) {
+    srtobject->thread =
+        g_thread_try_new ("GstSRTObjectListener", thread_func, srtobject, error);
 
-  if (*error != NULL) {
-    goto failed;
-  }
+    if (*error != NULL) {
+      goto failed;
+    }
 
-  return TRUE;
+    return TRUE;
+  }
 
 failed:
 

ext/srt/gstsrtobject.h

@@ -70,6 +70,8 @@ struct _GstSRTObject
 
   gboolean                     wait_for_connection;
 
+  gboolean                     authentication;
+
   guint64                      previous_bytes;
 };
 

ext/srt/gstsrtsink.c

@@ -58,6 +58,7 @@ enum
   SIG_CALLER_ADDED,
   SIG_CALLER_REMOVED,
   SIG_CALLER_REJECTED,
+  SIG_CALLER_CONNECTING,
 
   LAST_SIGNAL
 };
@@ -69,13 +70,36 @@ static void gst_srt_sink_uri_handler_init (gpointer g_iface,
 static gchar *gst_srt_sink_uri_get_uri (GstURIHandler * handler);
 static gboolean gst_srt_sink_uri_set_uri (GstURIHandler * handler,
     const gchar * uri, GError ** error);
+static gboolean caller_connecting (GstSRTSink *self,
+    GSocketAddress * addr, const gchar * username, gpointer data);
+static gboolean authentication_accumulator (GSignalInvocationHint * ihint,
+    GValue * return_accu, const GValue * handler_return, gpointer data);
 
 #define gst_srt_sink_parent_class parent_class
 G_DEFINE_TYPE_WITH_CODE (GstSRTSink, gst_srt_sink,
     GST_TYPE_BASE_SINK,
     G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_srt_sink_uri_handler_init)
     GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, "srtsink", 0, "SRT Sink"));
 
+static gboolean
+caller_connecting (GstSRTSink *self,
+    GSocketAddress * addr, const gchar * username, gpointer data)
+{
+  /* Accept all connections. */
+  return TRUE;
+}
+
+static gboolean
+authentication_accumulator (GSignalInvocationHint * ihint,
+    GValue * return_accu, const GValue * handler_return, gpointer data)
+{
+  gboolean ret = g_value_get_boolean (handler_return);
+  /* Handlers return TRUE on authentication success and we want to stop on
+   * the first failure. */
+  g_value_set_boolean (return_accu, ret);
+  return ret;
+}
+
 static void
 gst_srt_sink_set_property (GObject * object,
     guint prop_id, const GValue * value, GParamSpec * pspec)
@@ -312,6 +336,18 @@ gst_srt_sink_class_init (GstSRTSinkClass * klass)
       g_signal_new ("caller-rejected", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, 0, NULL, NULL, NULL, G_TYPE_NONE, 0);
 
+  /**
+   * GstSRTSink::caller-connecting:
+   * @gstsrtsink: the srtsink element that emitted this signal
+   *
+   * A caller's connection to srtsink.
+   */
+  signals[SIG_CALLER_CONNECTING] =
+      g_signal_new_class_handler ("caller-connecting", G_TYPE_FROM_CLASS (klass),
+      G_SIGNAL_RUN_LAST, G_CALLBACK (caller_connecting),
+      authentication_accumulator, NULL, NULL, G_TYPE_BOOLEAN,
+      2, G_TYPE_SOCKET_ADDRESS, G_TYPE_STRING);
+
   gst_srt_object_install_properties_helper (gobject_class);
 
   gst_element_class_add_static_pad_template (gstelement_class, &sink_template);

ext/srt/gstsrtsink.h

@@ -56,6 +56,8 @@ struct _GstSRTSinkClass {
 
   void (*caller_added)      (GstSRTSink *self, int sock, GSocketAddress * addr);
   void (*caller_removed)    (GstSRTSink *self, int sock, GSocketAddress * addr);
+  gboolean (*caller_connecting) (GstSRTSink *self, GSocketAddress * peer_address,
+	const gchar * stream_id, gpointer data);
 };
 
 GType gst_srt_sink_get_type (void);

ext/srt/gstsrtsrc.c

@@ -62,6 +62,7 @@ enum
   SIG_CALLER_ADDED,
   SIG_CALLER_REMOVED,
   SIG_CALLER_REJECTED,
+  SIG_CALLER_CONNECTING,
 
   LAST_SIGNAL
 };
@@ -73,13 +74,36 @@ static void gst_srt_src_uri_handler_init (gpointer g_iface,
 static gchar *gst_srt_src_uri_get_uri (GstURIHandler * handler);
 static gboolean gst_srt_src_uri_set_uri (GstURIHandler * handler,
     const gchar * uri, GError ** error);
+static gboolean src_caller_connecting (GstSRTSrc *self,
+    GSocketAddress * addr, const gchar * username, gpointer data);
+static gboolean src_authentication_accumulator (GSignalInvocationHint * ihint,
+    GValue * return_accu, const GValue * handler_return, gpointer data);
 
 #define gst_srt_src_parent_class parent_class
 G_DEFINE_TYPE_WITH_CODE (GstSRTSrc, gst_srt_src,
     GST_TYPE_PUSH_SRC,
     G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_srt_src_uri_handler_init)
     GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, "srtsrc", 0, "SRT Source"));
 
+static gboolean
+src_caller_connecting (GstSRTSrc *self,
+    GSocketAddress * addr, const gchar * username, gpointer data)
+{
+  /* Accept all connections. */
+  return TRUE;
+}
+
+static gboolean
+src_authentication_accumulator (GSignalInvocationHint * ihint,
+    GValue * return_accu, const GValue * handler_return, gpointer data)
+{
+  gboolean ret = g_value_get_boolean (handler_return);
+  /* Handlers return TRUE on authentication success and we want to stop on
+   * the first failure. */
+  g_value_set_boolean (return_accu, ret);
+  return ret;
+}
+
 static gboolean
 gst_srt_src_start (GstBaseSrc * bsrc)
 {
@@ -286,6 +310,18 @@ gst_srt_src_class_init (GstSRTSrcClass * klass)
       g_signal_new ("caller-rejected", G_TYPE_FROM_CLASS (klass),
       G_SIGNAL_RUN_LAST, 0, NULL, NULL, NULL, G_TYPE_NONE, 0);
 
+  /**
+   * GstSRTSrc::caller-connecting:
+   * @gstsrtsrc: the srtsrc element that emitted this signal
+   *
+   * A caller's connection to srtsrc.
+   */
+  signals[SIG_CALLER_CONNECTING] =
+      g_signal_new_class_handler ("caller-connecting", G_TYPE_FROM_CLASS (klass),
+      G_SIGNAL_RUN_LAST, G_CALLBACK (src_caller_connecting),
+      src_authentication_accumulator, NULL, NULL, G_TYPE_BOOLEAN,
+      2, G_TYPE_SOCKET_ADDRESS, G_TYPE_STRING);
+
   gst_srt_object_install_properties_helper (gobject_class);
 
   gst_element_class_add_static_pad_template (gstelement_class, &src_template);

ext/srt/gstsrtsrc.h

@@ -56,6 +56,8 @@ struct _GstSRTSrcClass {
 
   void (*caller_added)      (GstSRTSrc *self, int sock, GSocketAddress * addr);
   void (*caller_removed)    (GstSRTSrc *self, int sock, GSocketAddress * addr);
+  gboolean (*src_caller_connecting) (GstSRTSrc *self, GSocketAddress * peer_address,
+	const gchar * stream_id, gpointer data);
 };
 
 GType   gst_srt_src_get_type (void);