rtmp2: reimplement librtmp's connection parameters for the connect packet

librtmp allows for attaching arbitrary AMF objects to the end of the
connect packet, and this is commonly used for authenticating with
servers.

Add a new property, extra-connect-args, that mimics librtmp's behavior.

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

subprojects/gst-plugins-bad/docs/plugins/gst_plugins_cache.json

@@ -236393,6 +236393,18 @@
                         "type": "guint",
                         "writable": true
                     },
+                    "extra-connect-args": {
+                        "blurb": "librtmp-style arbitrary data to be appended to the \"connect\" command",
+                        "conditionally-available": false,
+                        "construct": false,
+                        "construct-only": false,
+                        "controllable": false,
+                        "default": "NULL",
+                        "mutable": "null",
+                        "readable": true,
+                        "type": "gchararray",
+                        "writable": true
+                    },
                     "peak-kbps": {
                         "blurb": "Bitrate in kbit/sec to pace outgoing packets",
                         "conditionally-available": false,
@@ -236472,6 +236484,18 @@
                         "type": "gboolean",
                         "writable": true
                     },
+                    "extra-connect-args": {
+                        "blurb": "librtmp-style arbitrary data to be appended to the \"connect\" command",
+                        "conditionally-available": false,
+                        "construct": false,
+                        "construct-only": false,
+                        "controllable": false,
+                        "default": "NULL",
+                        "mutable": "null",
+                        "readable": true,
+                        "type": "gchararray",
+                        "writable": true
+                    },
                     "idle-timeout": {
                         "blurb": "The maximum allowed time in seconds for valid packets not to arrive from the peer (0 = no timeout)",
                         "conditionally-available": false,

subprojects/gst-plugins-bad/gst/rtmp2/gstrtmp2sink.c

@@ -151,6 +151,7 @@ enum
   PROP_CHUNK_SIZE,
   PROP_STATS,
   PROP_STOP_COMMANDS,
+  PROP_EXTRA_CONNECT_ARGS,
 };
 
 /* pad templates */
@@ -247,6 +248,28 @@ gst_rtmp2_sink_class_init (GstRtmp2SinkClass * klass)
           GST_TYPE_RTMP_STOP_COMMANDS, GST_RTMP_DEFAULT_STOP_COMMANDS,
           (GParamFlags) (G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)));
 
+  /**
+   * GstRtmp2Sink:extra-connect-args:
+   *
+   * Parse and append librtmp-style arbitrary data to the "connect" command.
+   * It can be used for non-standard authentication with some servers.
+   *
+   * The format is a whitespace-separated series of "conn=type:data" strings.
+   * Valid types are:
+   *
+   * - "B" for boolean with "0" and "1" for false and true, respectively, e.g.
+   *   "conn=B:1".
+   * - "N" for numbers in double format, e.g. "conn=N:1.23".
+   * - "S" for strings, e.g. "conn=S:somepassword".
+   * - "O" for objects and "N"-prefixed named values are not yet supported.
+   *
+   * Since: 1.26
+   */
+  g_object_class_install_property (gobject_class, PROP_EXTRA_CONNECT_ARGS,
+      g_param_spec_string ("extra-connect-args", "librtmp-style arbitrary data",
+          "librtmp-style arbitrary data to be appended to the \"connect\" command",
+          NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
   gst_type_mark_as_plugin_api (GST_TYPE_RTMP_LOCATION_HANDLER, 0);
   GST_DEBUG_CATEGORY_INIT (gst_rtmp2_sink_debug_category, "rtmp2sink", 0,
       "debug category for rtmp2sink element");
@@ -386,6 +409,12 @@ gst_rtmp2_sink_set_property (GObject * object, guint property_id,
       self->stop_commands = g_value_get_flags (value);
       GST_OBJECT_UNLOCK (self);
       break;
+    case PROP_EXTRA_CONNECT_ARGS:
+      GST_OBJECT_LOCK (self);
+      g_free (self->location.extra_connect_args);
+      self->location.extra_connect_args = g_value_dup_string (value);
+      GST_OBJECT_UNLOCK (self);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
       break;
@@ -488,6 +517,11 @@ gst_rtmp2_sink_get_property (GObject * object, guint property_id,
       g_value_set_flags (value, self->stop_commands);
       GST_OBJECT_UNLOCK (self);
       break;
+    case PROP_EXTRA_CONNECT_ARGS:
+      GST_OBJECT_LOCK (self);
+      g_value_set_string (value, self->location.extra_connect_args);
+      GST_OBJECT_UNLOCK (self);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
       break;

subprojects/gst-plugins-bad/gst/rtmp2/gstrtmp2src.c

@@ -143,6 +143,7 @@ enum
   PROP_STATS,
   PROP_IDLE_TIMEOUT,
   PROP_NO_EOF_IS_ERROR,
+  PROP_EXTRA_CONNECT_ARGS,
 };
 
 #define DEFAULT_IDLE_TIMEOUT 0
@@ -237,6 +238,28 @@ gst_rtmp2_src_class_init (GstRtmp2SrcClass * klass)
           "If not set, those are reported using EOS", FALSE,
           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
+  /**
+   * GstRtmp2Src:extra-connect-args:
+   *
+   * Parse and append librtmp-style arbitrary data to the "connect" command.
+   * It can be used for non-standard authentication with some servers.
+   *
+   * The format is a whitespace-separated series of "conn=type:data" strings.
+   * Valid types are:
+   *
+   * - "B" for boolean with "0" and "1" for false and true, respectively, e.g.
+   *   "conn=B:1".
+   * - "N" for numbers in double format, e.g. "conn=N:1.23".
+   * - "S" for strings, e.g. "conn=S:somepassword".
+   * - "O" for objects and "N"-prefixed named values are not yet supported.
+   *
+   * Since: 1.26
+   */
+  g_object_class_install_property (gobject_class, PROP_EXTRA_CONNECT_ARGS,
+      g_param_spec_string ("extra-connect-args", "librtmp-style arbitrary data",
+          "librtmp-style arbitrary data to be appended to the \"connect\" command",
+          NULL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
   GST_DEBUG_CATEGORY_INIT (gst_rtmp2_src_debug_category, "rtmp2src", 0,
       "debug category for rtmp2src element");
 }
@@ -354,6 +377,12 @@ gst_rtmp2_src_set_property (GObject * object, guint property_id,
       self->no_eof_is_error = g_value_get_boolean (value);
       GST_OBJECT_UNLOCK (self);
       break;
+    case PROP_EXTRA_CONNECT_ARGS:
+      GST_OBJECT_LOCK (self);
+      g_free (self->location.extra_connect_args);
+      self->location.extra_connect_args = g_value_dup_string (value);
+      GST_OBJECT_UNLOCK (self);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
       break;
@@ -451,6 +480,11 @@ gst_rtmp2_src_get_property (GObject * object, guint property_id,
       g_value_set_boolean (value, self->no_eof_is_error);
       GST_OBJECT_UNLOCK (self);
       break;
+    case PROP_EXTRA_CONNECT_ARGS:
+      GST_OBJECT_LOCK (self);
+      g_value_set_string (value, self->location.extra_connect_args);
+      GST_OBJECT_UNLOCK (self);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
       break;

subprojects/gst-plugins-bad/gst/rtmp2/rtmp/amf.c

@@ -1205,3 +1205,36 @@ gst_amf_serialize_command_valist (gdouble transaction_id,
 
   return g_byte_array_free_to_bytes (array);
 }
+
+GBytes *
+gst_amf_serialize_command_with_args (gdouble transaction_id,
+    const gchar * command_name, gsize n_arguments,
+    const GstAmfNode ** arguments)
+{
+  GByteArray *array = g_byte_array_new ();
+  gsize i = 0;
+
+  g_return_val_if_fail (command_name, NULL);
+  g_return_val_if_fail (n_arguments, NULL);
+  g_return_val_if_fail (arguments, NULL);
+
+  init_static ();
+
+  GST_LOG ("Serializing command '%s', transid %.0f", command_name,
+      transaction_id);
+
+  serialize_u8 (array, GST_AMF_TYPE_STRING);
+  serialize_string (array, command_name, -1);
+  serialize_u8 (array, GST_AMF_TYPE_NUMBER);
+  serialize_number (array, transaction_id);
+
+  for (i = 0; i < n_arguments; i++) {
+    serialize_value (array, arguments[i]);
+    dump_argument (arguments[i], i);
+  }
+
+  GST_TRACE ("Done serializing; consumed %" G_GSIZE_FORMAT
+      "args and produced %u bytes", i, array->len);
+
+  return g_byte_array_free_to_bytes (array);
+}

subprojects/gst-plugins-bad/gst/rtmp2/rtmp/amf.h

@@ -112,6 +112,8 @@ GBytes * gst_amf_serialize_command (gdouble transaction_id,
     const gchar * command_name, const GstAmfNode * argument, ...) G_GNUC_NULL_TERMINATED;
 GBytes * gst_amf_serialize_command_valist (gdouble transaction_id,
     const gchar * command_name, const GstAmfNode * argument, va_list va_args);
+GBytes * gst_amf_serialize_command_with_args (gdouble transaction_id,
+    const gchar * command_name, gsize n_arguments, const GstAmfNode ** arguments);
 
 G_END_DECLS
 #endif

subprojects/gst-plugins-bad/gst/rtmp2/rtmp/rtmpclient.c

@@ -41,6 +41,15 @@ static void create_stream_done (const gchar * command_name, GPtrArray * args,
 static void on_publish_or_play_status (const gchar * command_name,
     GPtrArray * args, gpointer user_data);
 
+GQuark
+gst_rtmp_conn_parsing_error_quark (void)
+{
+  static GQuark quark = 0;
+  if (!quark)
+    quark = g_quark_from_static_string ("gst-rtmp-conn-parsing-error-quark");
+  return quark;
+}
+
 static void
 init_debug (void)
 {
@@ -200,6 +209,7 @@ gst_rtmp_location_copy (GstRtmpLocation * dest, const GstRtmpLocation * src)
   dest->username = g_strdup (src->username);
   dest->password = g_strdup (src->password);
   dest->secure_token = g_strdup (src->secure_token);
+  dest->extra_connect_args = g_strdup (src->extra_connect_args);
   dest->authmod = src->authmod;
   dest->timeout = src->timeout;
   dest->tls_flags = src->tls_flags;
@@ -219,6 +229,7 @@ gst_rtmp_location_clear (GstRtmpLocation * location)
   g_clear_pointer (&location->username, g_free);
   g_clear_pointer (&location->password, g_free);
   g_clear_pointer (&location->secure_token, g_free);
+  g_clear_pointer (&location->extra_connect_args, g_free);
   g_clear_pointer (&location->flash_ver, g_free);
   location->publish = FALSE;
 }
@@ -580,10 +591,151 @@ do_adobe_auth (const gchar * username, const gchar * password,
   return auth_query;
 }
 
+static GstAmfNode *
+parse_conn_token (gchar type, const gchar * value, GError ** error)
+{
+  g_return_val_if_fail (error == NULL || *error == NULL, FALSE);
+
+  // Function called without a type
+  g_assert (type);
+
+  gchar *end_ptr;
+  gboolean bool_;
+
+  GST_TRACE ("Parsing Connection token of Type: %c and Value: %s", type, value);
+
+  switch (type) {
+    case 'N':
+      // Empty value
+      if (value[0] == '\0') {
+        g_set_error (error,
+            GST_RTMP_CONN_PARSING_ERROR,
+            GST_RTMP_CONN_PARSING_ERROR_INVALID_VALUE,
+            "Found Numeric type, but the value is an empty string");
+        return NULL;
+      }
+
+      gdouble num = g_ascii_strtod (value, &end_ptr);
+      if (end_ptr[0] != '\0') {
+        g_set_error (error,
+            GST_RTMP_CONN_PARSING_ERROR,
+            GST_RTMP_CONN_PARSING_ERROR_FAILED_PARSING_DOUBLE,
+            "Failed to convert %s to double", value);
+        return NULL;
+      }
+
+      return gst_amf_node_new_number (num);
+    case 'S':
+      return gst_amf_node_new_string (value, -1);
+    case 'B':
+      // We are mimicking the behavior of librtmp here, which
+      // is using atoi and thus every invalid string is false
+      // https://salsa.debian.org/multimedia-team/rtmpdump/-/blob/a56abc82a99e8c4497a421d9dbc06e4544ade200/librtmp/rtmp.c#L632-634
+      bool_ = g_ascii_strtoull (value, &end_ptr, 10);
+      if (end_ptr[0] != '\0') {
+        return gst_amf_node_new_boolean (FALSE);
+      }
+
+      return gst_amf_node_new_boolean (bool_);
+    case 'Z':
+      return gst_amf_node_new_null ();
+    case 'O':
+      // Unimplemented for now
+      // Error: Unsupported
+      // O:1 Starts the object, then we parse the other conn= until O:0
+      // Then finish the object and serialize it
+      g_set_error (error,
+          GST_RTMP_CONN_PARSING_ERROR,
+          GST_RTMP_CONN_PARSING_ERROR_UNSUPPORTED,
+          "Objects are not yet supported");
+      return NULL;
+    default:
+      g_set_error (error,
+          GST_RTMP_CONN_PARSING_ERROR,
+          GST_RTMP_CONN_PARSING_ERROR_INVALID_TYPE,
+          "Invalid data type passed: %c", type);
+      return NULL;
+  }
+}
+
+// LIBRTMP(3) can append arbitrary data to the connection packet of RTMP.
+// It does so using a "connection" parameter appended after the url.
+// For a description of the format, see LIBRTMP(3) Connection Parameters
+//
+// Here we parse the conn= options and replicate the behavior for librtmp
+static gboolean
+parse_librtmp_style_conn_props (const gchar * connect_string, GPtrArray * array,
+    GError ** error)
+{
+  g_return_val_if_fail (error == NULL || *error == NULL, FALSE);
+
+  gchar **params;
+
+  // Split the string "conn=S:Foo conn=B:Bar"
+  params = g_strsplit (connect_string, "conn=", -1);
+
+  for (gsize i = 0; params[i]; i++) {
+    const gchar *param = g_strstrip (params[i]);
+
+    // Continue on empty string
+    if (param[0] == '\0') {
+      continue;
+    }
+    // Check for Named field of an object
+    // Example token: 'NS:Foo:Bar'
+    // The [0] byte will always be 'N' and the [2] must be the colon, which
+    // only occurs on named fields
+    if (param[0] == 'N' && param[1] != ':' && param[1] != '\0'
+        && param[2] == ':') {
+      // TODO: Error out if we had not found an object before
+
+      // TODO: split the value and create the AMF node
+      // then append it to the amf object, with the name
+      g_set_error (error,
+          GST_RTMP_CONN_PARSING_ERROR,
+          GST_RTMP_CONN_PARSING_ERROR_UNSUPPORTED,
+          "Objects are not yet supported");
+      g_strfreev (params);
+      return FALSE;
+    }
+
+    if (param[1] != ':') {
+      g_set_error (error,
+          GST_RTMP_CONN_PARSING_ERROR,
+          GST_RTMP_CONN_PARSING_ERROR_INVALID_VALUE,
+          "Parameter values are not separated by colon (:): %s",
+          connect_string);
+      g_strfreev (params);
+      return FALSE;
+    }
+    // Example token: 'S:Bar'
+    // [0] is the type prefix: 'S', 'B', etc
+    // [1] should always be a ':' separator
+    // [2] and is our arbitrary data value
+    const gchar type_ = param[0];
+    const gchar *value = &param[2];
+
+    GError *parse_error = NULL;
+    GstAmfNode *node = parse_conn_token (type_, value, &parse_error);
+    if (!node) {
+      g_strfreev (params);
+      g_propagate_error (error, parse_error);
+      return FALSE;
+    }
+
+    g_ptr_array_add (array, node);
+  };
+
+  g_strfreev (params);
+
+  return TRUE;
+}
+
 static void
 send_connect (GTask * task)
 {
   ConnectTaskData *data = g_task_get_task_data (task);
+  GPtrArray *arguments = g_ptr_array_new_with_free_func (gst_amf_node_free);
   GstAmfNode *node;
   const gchar *app, *flash_ver;
   gchar *uri, *appstr = NULL, *uristr = NULL;
@@ -679,11 +831,32 @@ send_connect (GTask * task)
      * XXX: libavformat sends "pageUrl" here, if provided. */
   }
 
-  gst_rtmp_connection_send_command (data->connection, send_connect_done,
-      task, 0, "connect", node, NULL);
+  g_ptr_array_add (arguments, node);
+
+  /* Parse librtmp style connect parameters */
+  if (data->location.extra_connect_args
+      && data->location.extra_connect_args[0] != '\0') {
+    GError *error = NULL;
+    gboolean conn_result =
+        parse_librtmp_style_conn_props (data->location.extra_connect_args,
+        arguments,
+        &error);
+
+    // Failed to parse the connect-args prop
+    if (!conn_result) {
+      g_task_return_new_error (task, error->domain, error->code,
+          "Failed to parse extra connection args: %s", error->message);
+      g_clear_error (&error);
+      goto out;
+    }
+  }
+
+  gst_rtmp_connection_send_command_with_args (data->connection,
+      send_connect_done, task, 0, "connect", arguments->len,
+      (const GstAmfNode **) arguments->pdata);
 
 out:
-  gst_amf_node_free (node);
+  g_ptr_array_free (arguments, TRUE);
   g_free (uri);
 }
 

subprojects/gst-plugins-bad/gst/rtmp2/rtmp/rtmpclient.h

@@ -91,6 +91,7 @@ typedef struct _GstRtmpLocation
   gchar *username;
   gchar *password;
   gchar *secure_token;
+  gchar *extra_connect_args;
   GstRtmpAuthmod authmod;
   gint timeout;
   GTlsCertificateFlags tls_flags;
@@ -126,5 +127,17 @@ gboolean gst_rtmp_client_start_play_finish (GstRtmpConnection * connection,
 void gst_rtmp_client_stop_publish (GstRtmpConnection * connection,
     const gchar * stream, const GstRtmpStopCommands stop_commands);
 
+typedef enum
+{
+  GST_RTMP_CONN_PARSING_ERROR_INVALID_TYPE,
+  GST_RTMP_CONN_PARSING_ERROR_INVALID_VALUE,
+  GST_RTMP_CONN_PARSING_ERROR_FAILED_PARSING_DOUBLE,
+  GST_RTMP_CONN_PARSING_ERROR_UNSUPPORTED,
+} GstRtmpConnParsingError;
+
+GQuark gst_rtmp_conn_parsing_error_quark (void);
+
+#define GST_RTMP_CONN_PARSING_ERROR gst_rtmp_conn_parsing_error_quark ()
+
 G_END_DECLS
 #endif

subprojects/gst-plugins-bad/gst/rtmp2/rtmp/rtmpconnection.c

@@ -1180,6 +1180,53 @@ gst_rtmp_connection_send_command (GstRtmpConnection * connection,
   return transaction_id;
 }
 
+guint
+gst_rtmp_connection_send_command_with_args (GstRtmpConnection * connection,
+    GstRtmpCommandCallback response_command, gpointer user_data,
+    guint32 stream_id, const gchar * command_name,
+    gsize n_arguments, const GstAmfNode ** arguments)
+{
+
+  GstBuffer *buffer;
+  gdouble transaction_id = 0;
+  GBytes *payload;
+  guint8 *data;
+  gsize size;
+
+  g_return_val_if_fail (GST_IS_RTMP_CONNECTION (connection), 0);
+
+  if (connection->thread != g_thread_self ()) {
+    GST_ERROR_OBJECT (connection, "Called from wrong thread");
+  }
+
+  GST_DEBUG_OBJECT (connection,
+      "Sending command '%s' on stream id %" G_GUINT32_FORMAT,
+      command_name, stream_id);
+
+  if (response_command) {
+    Transaction *t;
+
+    transaction_id = ++connection->transaction_count;
+
+    GST_LOG_OBJECT (connection, "Registering %s for transid %.0f",
+        GST_DEBUG_FUNCPTR_NAME (response_command), transaction_id);
+
+    t = transaction_new (transaction_id, response_command, user_data);
+
+    connection->transactions = g_list_append (connection->transactions, t);
+  }
+
+  payload = gst_amf_serialize_command_with_args (transaction_id,
+      command_name, n_arguments, arguments);
+
+  data = g_bytes_unref_to_data (payload, &size);
+  buffer = gst_rtmp_message_new_wrapped (GST_RTMP_MESSAGE_TYPE_COMMAND_AMF0,
+      3, stream_id, data, size);
+
+  gst_rtmp_connection_queue_message (connection, buffer);
+  return transaction_id;
+}
+
 void
 gst_rtmp_connection_expect_command (GstRtmpConnection * connection,
     GstRtmpCommandCallback response_command, gpointer user_data,

subprojects/gst-plugins-bad/gst/rtmp2/rtmp/rtmpconnection.h

@@ -89,6 +89,12 @@ void gst_rtmp_connection_request_window_size (GstRtmpConnection * connection,
 void gst_rtmp_connection_set_data_frame (GstRtmpConnection * connection,
     GstBuffer * buffer);
 
+guint
+gst_rtmp_connection_send_command_with_args (GstRtmpConnection * connection,
+    GstRtmpCommandCallback response_command, gpointer user_data,
+    guint32 stream_id, const gchar * command_name,
+    gsize n_arguments, const GstAmfNode ** arguments);
+
 GstStructure * gst_rtmp_connection_get_null_stats (void);
 GstStructure * gst_rtmp_connection_get_stats (GstRtmpConnection * connection);