gst/rtmp2/rtmp/rtmphandshake.c

   1 /* GStreamer RTMP Library
   2  * Copyright (C) 2017 Make.TV, Inc. <info@make.tv>
   3  *   Contact: Jan Alexander Steffens (heftig) <jsteffens@make.tv>
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Library General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Library General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Library General Public
  16  * License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  18  * Boston, MA 02110-1301, USA.
  19  */
  20
  21 #ifdef HAVE_CONFIG_H
  22 #include "config.h"
  23 #endif
  24
  25 #include "rtmphandshake.h"
  26 #include "rtmputils.h"
  27
  28 #include <gst/gst.h>
  29 #include <string.h>
  30
  31 GST_DEBUG_CATEGORY_STATIC (gst_rtmp_handshake_debug_category);
  32 #define GST_CAT_DEFAULT gst_rtmp_handshake_debug_category
  33
  34 static void
  35 init_debug (void)
  36 {
  37   static volatile gsize done = 0;
  38   if (g_once_init_enter (&done)) {
  39     GST_DEBUG_CATEGORY_INIT (gst_rtmp_handshake_debug_category, "rtmphandshake",
  40         0, "debug category for the rtmp connection handshake");
  41     g_once_init_leave (&done, 1);
  42   }
  43 }
  44
  45 static void client_handshake1_done (GObject * source, GAsyncResult * result,
  46     gpointer user_data);
  47 static void client_handshake2_done (GObject * source, GAsyncResult * result,
  48     gpointer user_data);
  49 static void client_handshake3_done (GObject * source, GAsyncResult * result,
  50     gpointer user_data);
  51
  52 static inline void
  53 serialize_u8 (GByteArray * array, guint8 value)
  54 {
  55   g_byte_array_append (array, (guint8 *) & value, sizeof value);
  56 }
  57
  58 static inline void
  59 serialize_u32 (GByteArray * array, guint32 value)
  60 {
  61   value = GUINT32_TO_BE (value);
  62   g_byte_array_append (array, (guint8 *) & value, sizeof value);
  63 }
  64
  65 #define SIZE_P0 1
  66 #define SIZE_P1 1536
  67 #define SIZE_P2 SIZE_P1
  68 #define SIZE_P0P1 (SIZE_P0 + SIZE_P1)
  69 #define SIZE_P0P1P2 (SIZE_P0P1 + SIZE_P2)
  70 #define SIZE_RANDOM (SIZE_P1 - 8)
  71
  72 typedef struct
  73 {
  74   GBytes *random_bytes;
  75   gboolean strict;
  76 } HandshakeData;
  77
  78 static GBytes *
  79 handshake_random_data (void)
  80 {
  81   G_STATIC_ASSERT (SIZE_RANDOM % 4 == 0);
  82
  83   GByteArray *ba = g_byte_array_sized_new (SIZE_RANDOM);
  84   gint i;
  85
  86   for (i = 0; i < SIZE_RANDOM; i += 4) {
  87     serialize_u32 (ba, g_random_int ());
  88   }
  89
  90   return g_byte_array_free_to_bytes (ba);
  91 }
  92
  93 static HandshakeData *
  94 handshake_data_new (gboolean strict)
  95 {
  96   HandshakeData *data = g_slice_new0 (HandshakeData);
  97   data->random_bytes = handshake_random_data ();
  98   data->strict = strict;
  99   return data;
 100 }
 101
 102 static void
 103 handshake_data_free (gpointer ptr)
 104 {
 105   HandshakeData *data = ptr;
 106   g_clear_pointer (&data->random_bytes, g_bytes_unref);
 107   g_slice_free (HandshakeData, data);
 108 }
 109
 110 static gboolean
 111 handshake_data_check (HandshakeData * data, const guint8 * p2)
 112 {
 113   const guint8 *ourrandom = g_bytes_get_data (data->random_bytes, NULL);
 114   return memcmp (ourrandom, p2 + 8, SIZE_P2 - 8) == 0;
 115 }
 116
 117 static GBytes *
 118 create_c0c1 (GBytes * random_bytes)
 119 {
 120   GByteArray *ba = g_byte_array_sized_new (SIZE_P0P1);
 121
 122   /* C0 version */
 123   serialize_u8 (ba, 3);
 124
 125   /* C1 time */
 126   serialize_u32 (ba, g_get_monotonic_time () / 1000);
 127
 128   /* C1 zero */
 129   serialize_u32 (ba, 0);
 130
 131   /* C1 random data */
 132   gst_rtmp_byte_array_append_bytes (ba, random_bytes);
 133
 134   GST_DEBUG ("Sending C0+C1");
 135   GST_MEMDUMP (">>> C0", ba->data, SIZE_P0);
 136   GST_MEMDUMP (">>> C1", ba->data + SIZE_P0, SIZE_P1);
 137
 138   return g_byte_array_free_to_bytes (ba);
 139 }
 140
 141 void
 142 gst_rtmp_client_handshake (GIOStream * stream, gboolean strict,
 143     GCancellable * cancellable, GAsyncReadyCallback callback,
 144     gpointer user_data)
 145 {
 146   GTask *task;
 147   HandshakeData *data;
 148
 149   g_return_if_fail (G_IS_IO_STREAM (stream));
 150
 151   init_debug ();
 152   GST_INFO ("Starting client handshake");
 153
 154   task = g_task_new (stream, cancellable, callback, user_data);
 155   data = handshake_data_new (strict);
 156   g_task_set_task_data (task, data, handshake_data_free);
 157
 158   {
 159     GOutputStream *os = g_io_stream_get_output_stream (stream);
 160     GBytes *bytes = create_c0c1 (data->random_bytes);
 161
 162     gst_rtmp_output_stream_write_all_bytes_async (os,
 163         bytes, G_PRIORITY_DEFAULT,
 164         g_task_get_cancellable (task), client_handshake1_done, task);
 165
 166     g_bytes_unref (bytes);
 167   }
 168 }
 169
 170 static void
 171 client_handshake1_done (GObject * source, GAsyncResult * result,
 172     gpointer user_data)
 173 {
 174   GOutputStream *os = G_OUTPUT_STREAM (source);
 175   GTask *task = user_data;
 176   GIOStream *stream = g_task_get_source_object (task);
 177   GInputStream *is = g_io_stream_get_input_stream (stream);
 178   GError *error = NULL;
 179   gboolean res;
 180
 181   res = gst_rtmp_output_stream_write_all_bytes_finish (os, result, &error);
 182   if (!res) {
 183     GST_ERROR ("Failed to send C0+C1: %s", error->message);
 184     g_task_return_error (task, error);
 185     g_object_unref (task);
 186     return;
 187   }
 188
 189   GST_DEBUG ("Sent C0+C1, waiting for S0+S1+S2");
 190   gst_rtmp_input_stream_read_all_bytes_async (is, SIZE_P0P1P2,
 191       G_PRIORITY_DEFAULT, g_task_get_cancellable (task),
 192       client_handshake2_done, task);
 193 }
 194
 195 static GBytes *
 196 create_c2 (const guint8 * s0s1s2)
 197 {
 198   G_STATIC_ASSERT (SIZE_P1 == SIZE_P2);
 199
 200   GByteArray *ba = g_byte_array_sized_new (SIZE_P2);
 201   gint64 c2time = g_get_monotonic_time ();
 202
 203   /* Copy S1 to C2 */
 204   g_byte_array_set_size (ba, SIZE_P2);
 205   memcpy (ba->data, s0s1s2 + SIZE_P0, SIZE_P1);
 206
 207   /* C2 time2 */
 208   GST_WRITE_UINT32_BE (ba->data + 4, c2time / 1000);
 209
 210   GST_DEBUG ("Sending C2");
 211   GST_MEMDUMP (">>> C2", ba->data, SIZE_P2);
 212
 213   return g_byte_array_free_to_bytes (ba);
 214 }
 215
 216 static void
 217 client_handshake2_done (GObject * source, GAsyncResult * result,
 218     gpointer user_data)
 219 {
 220   GInputStream *is = G_INPUT_STREAM (source);
 221   GTask *task = user_data;
 222   GIOStream *stream = g_task_get_source_object (task);
 223   HandshakeData *data = g_task_get_task_data (task);
 224   GError *error = NULL;
 225   GBytes *res;
 226   const guint8 *s0s1s2;
 227   gsize size;
 228
 229   res = gst_rtmp_input_stream_read_all_bytes_finish (is, result, &error);
 230   if (!res) {
 231     GST_ERROR ("Failed to read S0+S1+S2: %s", error->message);
 232     g_task_return_error (task, error);
 233     g_object_unref (task);
 234     return;
 235   }
 236
 237   s0s1s2 = g_bytes_get_data (res, &size);
 238   if (size < SIZE_P0P1P2) {
 239     GST_ERROR ("Short read (want %d have %" G_GSIZE_FORMAT ")", SIZE_P0P1P2,
 240         size);
 241     g_task_return_new_error (task, G_IO_ERROR, G_IO_ERROR_PARTIAL_INPUT,
 242         "Short read (want %d have %" G_GSIZE_FORMAT ")", SIZE_P0P1P2, size);
 243     g_object_unref (task);
 244     goto out;
 245   }
 246
 247   GST_DEBUG ("Got S0+S1+S2");
 248   GST_MEMDUMP ("<<< S0", s0s1s2, SIZE_P0);
 249   GST_MEMDUMP ("<<< S1", s0s1s2 + SIZE_P0, SIZE_P1);
 250   GST_MEMDUMP ("<<< S2", s0s1s2 + SIZE_P0P1, SIZE_P2);
 251
 252   if (handshake_data_check (data, s0s1s2 + SIZE_P0P1)) {
 253     GST_DEBUG ("S2 random data matches C1");
 254   } else {
 255     if (data->strict) {
 256       GST_ERROR ("Handshake response data did not match");
 257       g_task_return_new_error (task, G_IO_ERROR, G_IO_ERROR_INVALID_DATA,
 258           "Handshake response data did not match");
 259       g_object_unref (task);
 260       goto out;
 261     }
 262
 263     GST_WARNING ("Handshake reponse data did not match; continuing anyway");
 264   }
 265
 266   {
 267     GOutputStream *os = g_io_stream_get_output_stream (stream);
 268     GBytes *bytes = create_c2 (s0s1s2);
 269
 270     gst_rtmp_output_stream_write_all_bytes_async (os,
 271         bytes, G_PRIORITY_DEFAULT,
 272         g_task_get_cancellable (task), client_handshake3_done, task);
 273
 274     g_bytes_unref (bytes);
 275   }
 276
 277 out:
 278   g_bytes_unref (res);
 279 }
 280
 281 static void
 282 client_handshake3_done (GObject * source, GAsyncResult * result,
 283     gpointer user_data)
 284 {
 285   GOutputStream *os = G_OUTPUT_STREAM (source);
 286   GTask *task = user_data;
 287   GError *error = NULL;
 288   gboolean res;
 289
 290   res = gst_rtmp_output_stream_write_all_bytes_finish (os, result, &error);
 291   if (!res) {
 292     GST_ERROR ("Failed to send C2: %s", error->message);
 293     g_task_return_error (task, error);
 294     g_object_unref (task);
 295     return;
 296   }
 297
 298   GST_DEBUG ("Sent C2");
 299   GST_INFO ("Client handshake finished");
 300
 301   g_task_return_boolean (task, TRUE);
 302   g_object_unref (task);
 303 }
 304
 305 gboolean
 306 gst_rtmp_client_handshake_finish (GIOStream * stream, GAsyncResult * result,
 307     GError ** error)
 308 {
 309   g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
 310   return g_task_propagate_boolean (G_TASK (result), error);
 311 }