feat(device): add ValentPacketInputStream

Add a packet reading stream, both for convenience and to handle
untrusted sources that may attempt DoS attacks by overfilling
the input buffer.

When a source is verified, the stream can be marked as trusted,
allowing the buffer to expand as necessary.

Author: andyholmes

src/libvalent/device/meson.build

@@ -23,6 +23,7 @@ libvalent_device_public_headers = [
 libvalent_device_private_headers = [
   'valent-device-impl.h',
   'valent-device-private.h',
+  'valent-packet-input-stream.h',
 ]
 
 libvalent_device_enum_headers = [
@@ -46,6 +47,7 @@ libvalent_device_public_sources = [
   'valent-device-plugin.c',
   'valent-device-transfer.c',
   'valent-packet.c',
+  'valent-packet-input-stream.c',
 ]
 
 

src/libvalent/device/valent-channel.c

@@ -13,6 +13,7 @@
 
 #include "valent-certificate.h"
 #include "valent-packet.h"
+#include "valent-packet-input-stream.h"
 
 #include "valent-channel.h"
 
@@ -55,7 +56,7 @@ typedef struct
   JsonNode         *peer_identity;
 
   /* Packet Buffer */
-  GDataInputStream *input_buffer;
+  ValentPacketInputStream *input_buffer;
   GMainLoop        *output_buffer;
 } ValentChannelPrivate;
 
@@ -278,10 +279,13 @@ valent_channel_set_base_stream (ValentChannel *self,
       g_autoptr (GThread) thread = NULL;
       g_autoptr (GError) error = NULL;
 
+      // FIXME: associate the device paired state with the channel trust
       input_stream = g_io_stream_get_input_stream (base_stream);
-      priv->input_buffer = g_object_new (G_TYPE_DATA_INPUT_STREAM,
+      priv->input_buffer = g_object_new (VALENT_TYPE_PACKET_INPUT_STREAM,
                                          "base-stream",       input_stream,
+                                         "buffer-size",       VALENT_PACKET_DEFAULT_BUFFER_SIZE,
                                          "close-base-stream", FALSE,
+                                         "trusted",           TRUE,
                                          NULL);
 
       context = g_main_context_new ();
@@ -761,34 +765,15 @@ valent_channel_close_finish (ValentChannel  *channel,
 }
 
 static void
-valent_channel_read_packet_cb (GObject      *object,
-                               GAsyncResult *result,
-                               gpointer      user_data)
+valent_channel_read_packet_cb (ValentPacketInputStream *stream,
+                               GAsyncResult            *result,
+                               gpointer                 user_data)
 {
   g_autoptr (GTask) task = G_TASK (g_steal_pointer (&user_data));
-  g_autofree char *line = NULL;
   JsonNode *packet = NULL;
   GError *error = NULL;
 
-  line = g_data_input_stream_read_line_finish_utf8 (G_DATA_INPUT_STREAM (object),
-                                                    result,
-                                                    NULL,
-                                                    &error);
-  if (error != NULL)
-    {
-      g_task_return_error (task, g_steal_pointer (&error));
-      return;
-    }
-  else if (line == NULL)
-    {
-      g_task_return_new_error (task,
-                               G_IO_ERROR,
-                               G_IO_ERROR_CONNECTION_CLOSED,
-                               "Channel is closed");
-      return;
-    }
-
-  packet = valent_packet_deserialize (line, &error);
+  packet = valent_packet_input_stream_read_packet_finish (stream, result, &error);
   if (packet == NULL)
     {
       g_task_return_error (task, g_steal_pointer (&error));
@@ -832,11 +817,10 @@ valent_channel_read_packet (ValentChannel       *channel,
 
   if (!valent_channel_return_error_if_closed (channel, task))
     {
-      g_data_input_stream_read_line_async (priv->input_buffer,
-                                           G_PRIORITY_DEFAULT,
-                                           cancellable,
-                                           valent_channel_read_packet_cb,
-                                           g_object_ref (task));
+      valent_packet_input_stream_read_packet_async (priv->input_buffer,
+                                                    cancellable,
+                                                    (GAsyncReadyCallback)valent_channel_read_packet_cb,
+                                                    g_object_ref (task));
       valent_object_unlock (VALENT_OBJECT (channel));
     }