SCTP: refactor to fix encoder use-after-free bug

Because of the nature of not having a clear ownership between the association
and the elements (encoder or decoder) is not possible to rely on the
associations signals to have a valid reference on the encoder.
This also applies for the decoder.

Since g_signal_connect_object is not thread-safe (more on this in the documentation)
and we can't hold the association mutex while emitting a signal, it is
completely possible -as our stress tests demonstrate- that we ran into data
races.

To fix this situation, I introduce proper callback structures per element
context and inside the association we always keep a reference to the element
while calling the related callback.

subprojects/gst-plugins-bad/ext/sctp/gstsctpdec.c

@@ -163,9 +163,9 @@ static void gst_sctp_data_srcpad_loop (GstPad * pad);
 static gboolean configure_association (GstSctpDec * self);
 static void cleanup_association (GstSctpDec * self);
 static void on_gst_sctp_association_stream_reset (GstSctpAssociation *
-    gst_sctp_association, guint16 stream_id, GstSctpDec * self);
+    gst_sctp_association, guint16 stream_id, gpointer user_data);
 static void on_gst_sctp_association_restart (GstSctpAssociation *
-    gst_sctp_association, GstSctpDec * self);
+    gst_sctp_association, gpointer user_data);
 static void on_receive (GstSctpAssociation * gst_sctp_association,
     guint8 * buf, gsize length, guint16 stream_id, guint ppid,
     gpointer user_data);
@@ -528,6 +528,7 @@ static gboolean
 configure_association (GstSctpDec * self)
 {
   gint state;
+  GstSctpAssociationDecoderCtx ctx;
 
   GST_SCTP_DEC_ASSOC_MUTEX_LOCK (self);
   self->sctp_association = gst_sctp_association_get (self->sctp_association_id);
@@ -542,19 +543,14 @@ configure_association (GstSctpDec * self)
     return FALSE;
   }
 
-  self->signal_handler_stream_reset =
-      g_signal_connect_object (self->sctp_association, "stream-reset",
-      G_CALLBACK (on_gst_sctp_association_stream_reset), self, 0);
-
-  self->signal_handler_association_restart =
-      g_signal_connect_object (self->sctp_association, "association-restart",
-      G_CALLBACK (on_gst_sctp_association_restart), self, 0);
-
   g_object_set (self->sctp_association, "local-port", self->local_sctp_port,
       NULL);
 
-  gst_sctp_association_set_on_packet_received (self->sctp_association,
-      on_receive, gst_object_ref (self), gst_object_unref);
+  ctx.element = self;
+  ctx.stream_reset_cb = on_gst_sctp_association_stream_reset;
+  ctx.restart_cb = on_gst_sctp_association_restart;
+  ctx.packet_received_cb = on_receive;
+  gst_sctp_association_set_decoder_ctx (self->sctp_association, &ctx);
 
   GST_SCTP_DEC_ASSOC_MUTEX_UNLOCK (self);
 
@@ -707,10 +703,11 @@ get_pad_for_stream_id (GstSctpDec * self, guint16 stream_id)
 
 static void
 on_gst_sctp_association_stream_reset (GstSctpAssociation * gst_sctp_association,
-    guint16 stream_id, GstSctpDec * self)
+    guint16 stream_id, gpointer user_data)
 {
   gchar *pad_name;
   GstPad *srcpad;
+  GstSctpDec *self = user_data;
 
   GST_DEBUG_OBJECT (self, "Stream %u reset", stream_id);
 
@@ -727,8 +724,9 @@ on_gst_sctp_association_stream_reset (GstSctpAssociation * gst_sctp_association,
 
 static void
 on_gst_sctp_association_restart (GstSctpAssociation * gst_sctp_association,
-    GstSctpDec * self)
+    gpointer user_data)
 {
+  GstSctpDec *self = user_data;
   (void) gst_sctp_association;
   g_signal_emit (self, signals[SIGNAL_ASSOC_RESTART], 0);
 }
@@ -780,26 +778,16 @@ on_receive (GstSctpAssociation * sctp_association, guint8 * buf,
 static void
 cleanup_association (GstSctpDec * self)
 {
+  GstSctpAssociationDecoderCtx ctx;
+
   GST_SCTP_DEC_ASSOC_MUTEX_LOCK (self);
   if (!self->sctp_association) {
     GST_SCTP_DEC_ASSOC_MUTEX_UNLOCK (self);
     return;
   }
-
-  if (self->signal_handler_association_restart != 0) {
-    g_signal_handler_disconnect (self->sctp_association,
-        self->signal_handler_association_restart);
-    self->signal_handler_association_restart = 0;
-  }
-
-  if (self->signal_handler_stream_reset != 0) {
-    g_signal_handler_disconnect (self->sctp_association,
-        self->signal_handler_stream_reset);
-    self->signal_handler_stream_reset = 0;
-  }
-
-  gst_sctp_association_set_on_packet_received (self->sctp_association, NULL,
-      NULL, NULL);
+
+  memset (&ctx, 0, sizeof (GstSctpAssociationDecoderCtx));
+  gst_sctp_association_set_decoder_ctx (self->sctp_association, &ctx);
   gst_sctp_association_force_close (self->sctp_association);
   gst_sctp_association_unref (self->sctp_association);
   self->sctp_association = NULL;

subprojects/gst-plugins-bad/ext/sctp/gstsctpdec.h

@@ -55,8 +55,6 @@ struct _GstSctpDec
   guint local_sctp_port;
 
   GstSctpAssociation *sctp_association;
-  gulong signal_handler_stream_reset;
-  gulong signal_handler_association_restart;
 };
 
 struct _GstSctpDecClass

subprojects/gst-plugins-bad/ext/sctp/gstsctpenc.c

@@ -165,7 +165,7 @@ static gboolean
 gst_sctp_enc_src_activate_mode (GstPad * pad, GstObject * parent,
     GstPadMode mode, gboolean active);
 static void on_sctp_association_state_changed (GstSctpAssociation *
-    sctp_association, GParamSpec * pspec, GstSctpEnc * self);
+    sctp_association, GstSctpAssociationState state, gpointer user_data);
 
 static gboolean configure_association (GstSctpEnc * self);
 static void cleanup_association (GstSctpEnc * self);
@@ -998,6 +998,7 @@ static gboolean
 configure_association (GstSctpEnc * self)
 {
   gint state;
+  GstSctpAssociationEncoderCtx ctx;
 
   GST_SCTP_ENC_ASSOC_MUTEX_LOCK (self);
   self->sctp_association = gst_sctp_association_get (self->sctp_association_id);
@@ -1012,29 +1013,28 @@ configure_association (GstSctpEnc * self)
     return FALSE;
   }
 
-  self->signal_handler_state_changed =
-      g_signal_connect_object (self->sctp_association, "notify::state",
-      G_CALLBACK (on_sctp_association_state_changed), self, 0);
-
   g_object_set (self->sctp_association, "remote-port", self->remote_sctp_port,
       "use-sock-stream", self->use_sock_stream, "aggressive-heartbeat",
       self->aggressive_heartbeat, NULL);
 
-  gst_sctp_association_set_on_packet_out (self->sctp_association,
-      on_sctp_packet_out, gst_object_ref (self), gst_object_unref);
+  ctx.element = self;
+  ctx.state_change_cb = on_sctp_association_state_changed;
+  ctx.packet_out_cb = on_sctp_packet_out;
+  gst_sctp_association_set_encoder_ctx (self->sctp_association, &ctx);
+
   GST_SCTP_ENC_ASSOC_MUTEX_UNLOCK (self);
 
   return TRUE;
 }
 
 static void
-on_sctp_association_state_changed (GstSctpAssociation * sctp_association,
-    GParamSpec * pspec, GstSctpEnc * self)
+on_sctp_association_state_changed (GstSctpAssociation *
+    sctp_association, GstSctpAssociationState state, gpointer user_data)
 {
-  gint state;
-
-  g_object_get (sctp_association, "state", &state, NULL);
+  GstSctpEnc *self = (GstSctpEnc *) user_data;
 
+  /* we demand to have a valid encoder here */
+  g_assert (self);
   GST_DEBUG_OBJECT (self, "Association state changed to %d", state);
 
   switch (state) {
@@ -1116,20 +1116,16 @@ on_sctp_packet_out (GstSctpAssociation * _association, const guint8 * buf,
 static void
 cleanup_association (GstSctpEnc * self)
 {
+  GstSctpAssociationEncoderCtx ctx;
+
   GST_SCTP_ENC_ASSOC_MUTEX_LOCK (self);
   if (!self->sctp_association) {
     GST_SCTP_ENC_ASSOC_MUTEX_UNLOCK (self);
     return;
   }
 
-  if (self->signal_handler_state_changed != 0) {
-    g_signal_handler_disconnect (self->sctp_association,
-        self->signal_handler_state_changed);
-    self->signal_handler_state_changed = 0;
-  }
-
-  gst_sctp_association_set_on_packet_out (self->sctp_association, NULL, NULL,
-      NULL);
+  memset (&ctx, 0, sizeof (GstSctpAssociationEncoderCtx));
+  gst_sctp_association_set_encoder_ctx (self->sctp_association, &ctx);
   gst_sctp_association_force_close (self->sctp_association);
   gst_sctp_association_unref (self->sctp_association);
   self->sctp_association = NULL;

subprojects/gst-plugins-bad/ext/sctp/gstsctpenc.h

@@ -60,8 +60,6 @@ struct _GstSctpEnc
   GstDataQueue *outbound_sctp_packet_queue;
 
   GQueue pending_pads;
-
-  gulong signal_handler_state_changed;
 };
 
 struct _GstSctpEncClass