Merge pull request #3845 from matt335672/rdpedyc_data_fragments

dechunker: Add support for dynamic virtual channels.

common/dechunker.c

@@ -57,11 +57,17 @@ struct vc_dechunker
     struct stream *reassembly_s;
 };
 
+struct dyn_dechunker
+{
+    char name[64];
+    struct stream *reassembly_s;
+};
+
 enum
 {
     // This is a rather arbitrary figure, but one we are unlikely to
     // go below.  It's a sanity check for vc_dechunker_init()
-    E_MAX_CHUNK_SIZE_LOWER_LIMIT = 50
+    E_MAX_VC_CHUNK_SIZE_LOWER_LIMIT = 50
 };
 /*****************************************************************************/
 struct vc_dechunker *
@@ -72,7 +78,7 @@ vc_dechunker_init(const char *chan_name, int max_chunk_size)
     {
         LOG(LOG_LEVEL_ERROR, "vc_dechunker_init() called with no channel name");
     }
-    else if (max_chunk_size < E_MAX_CHUNK_SIZE_LOWER_LIMIT)
+    else if (max_chunk_size < E_MAX_VC_CHUNK_SIZE_LOWER_LIMIT)
     {
         LOG(LOG_LEVEL_ERROR, "Dechunker: Max chunk size for %s is too small",
             chan_name);
@@ -237,6 +243,9 @@ handle_reading_state(struct vc_dechunker *self,
                     // Tell the caller the stream is available.
                     self->state = E_DATA;
                     rv = E_VC_READY;
+                    LOG_DEVEL(LOG_LEVEL_INFO,
+                              "Dechunker: Reassembled PDU of size %d on %s",
+                              self->reassembly_s->size, self->name);
                 }
                 else
                 {
@@ -369,3 +378,172 @@ vc_dechunker_get_stream(struct vc_dechunker *self)
 
     return s;
 }
+/*****************************************************************************/
+struct dyn_dechunker *
+dyn_dechunker_init(const char *chan_name)
+{
+    struct dyn_dechunker *self = NULL;
+    if (chan_name == NULL)
+    {
+        LOG(LOG_LEVEL_ERROR,
+            "dyn_dechunker_init() called with no channel name");
+    }
+    else if ((self = g_new(struct dyn_dechunker, 1)) == NULL)
+    {
+        LOG(LOG_LEVEL_ERROR, "Dechunker: no memory for %s", chan_name);
+    }
+    else
+    {
+        strlcpy(self->name, chan_name, sizeof(self->name));
+        self->reassembly_s = NULL;
+    }
+
+    return self;
+}
+
+/*****************************************************************************/
+void
+dyn_dechunker_free(struct dyn_dechunker *self)
+{
+    if (self != NULL)
+    {
+        free_stream(self->reassembly_s);
+        free(self);
+    }
+}
+
+/*****************************************************************************/
+enum dyn_dechunker_status
+dyn_dechunker_process_first_chunk(struct dyn_dechunker *self,
+                                  struct stream *s, int total_size)
+{
+    enum dyn_dechunker_status status = E_DYN_ERROR;
+
+    int frag_size = s ? s_rem(s) : 0;
+    if (self == NULL || s == NULL)
+    {
+        ; // Nothing to be done
+    }
+    else if (total_size <= 1590 || frag_size > total_size)
+    {
+        // See [MS-RDPEDYC] 2.2.3
+        LOG(LOG_LEVEL_ERROR,
+            "Badly sized DYNVC_DATA_FIRST PDU received on dynamic channel %s",
+            self->name);
+    }
+    else if (self->reassembly_s != NULL)
+    {
+        LOG(LOG_LEVEL_ERROR,
+            "unexpected DYNVC_DATA_FIRST received on dynamic channel %s",
+            self->name);
+    }
+    else if (frag_size == total_size)
+    {
+        // This chunk contains all the data
+        status = E_DYN_INLINE_CHUNK;
+    }
+    else
+    {
+        make_stream(self->reassembly_s);
+        if (self->reassembly_s)
+        {
+            init_stream(self->reassembly_s, total_size);
+        }
+        if (self->reassembly_s == NULL || self->reassembly_s->data == NULL)
+        {
+            LOG(LOG_LEVEL_ERROR,
+                "Out of memory for dynamic PDU reassembly on %s",
+                self->name);
+        }
+        else
+        {
+            out_uint8p(self->reassembly_s, s->p, frag_size);
+            in_uint8s(s, frag_size);
+            status = E_DYN_IN_PROGRESS;
+        }
+    }
+
+    return status;
+}
+
+/*****************************************************************************/
+enum dyn_dechunker_status
+dyn_dechunker_process_data_chunk(struct dyn_dechunker *self,
+                                 struct stream *s)
+{
+    enum dyn_dechunker_status rv;
+
+    if (self == NULL || s == NULL)
+    {
+        rv = E_DYN_ERROR;
+    }
+    else if (self->reassembly_s == NULL)
+    {
+        rv = E_DYN_INLINE_CHUNK;
+    }
+    else
+    {
+        int frag_size = s_rem(s);
+        // We're currently reconstructing a data PDU from fragments
+        if (!s_check_rem_out(self-> reassembly_s, frag_size))
+        {
+            LOG(LOG_LEVEL_ERROR,
+                "Oversized DYNVC_DATA when reconstructing PDU on %s",
+                self->name);
+            rv = E_DYN_ERROR;
+        }
+        else
+        {
+            out_uint8p(self->reassembly_s, s->p, frag_size);
+            in_uint8s(s, frag_size);
+
+            if (s_rem_out(self->reassembly_s) == 0)
+            {
+                // Finished defragging
+                s_mark_end(self->reassembly_s);
+                self->reassembly_s->p = self->reassembly_s->data;
+                rv = E_DYN_READY;
+                LOG_DEVEL(LOG_LEVEL_INFO,
+                          "Dechunker: Reassembled PDU of size %d on %s",
+                          self->reassembly_s->size, self->name);
+            }
+            else
+            {
+                rv = E_DYN_IN_PROGRESS;
+            }
+        }
+    }
+
+    return rv;
+}
+
+/*****************************************************************************/
+struct stream *
+dyn_dechunker_get_stream(struct dyn_dechunker *self)
+{
+    struct stream *s;
+    const char *name = (self != NULL) ? self->name : "<unknown>";
+    if (self == NULL || self->reassembly_s == NULL ||
+            self->reassembly_s->end == self->reassembly_s->data)
+    {
+        LOG (LOG_LEVEL_ERROR,
+             "Dechunker: get stream called for %s with no data available",
+             name);
+        s = NULL;
+    }
+    else
+    {
+        // Pass ownership of the stream to the caller
+        s = self->reassembly_s;
+        self->reassembly_s = NULL; // So we don't free it ourselves!
+    }
+
+    return s;
+}
+
+/*****************************************************************************/
+int
+dyn_dechunker_pending(struct dyn_dechunker *self)
+{
+    return (self != NULL && self->reassembly_s != NULL);
+}

common/dechunker.h

@@ -34,11 +34,12 @@
 
 struct stream;
 
-/* Private type */
+/* Private types */
-struct vc_dechunker;
+struct vc_dechunker;   // static virtual channel dechunker
+struct dyn_dechunker;  // dynamic channel dechunker
 
 /**
- * Returned from dechunker_process_vc_chunk()
+ * Returned from vc_dechunker_process_chunk()
  */
 enum vc_dechunker_status
 {
@@ -49,7 +50,19 @@ enum vc_dechunker_status
 };
 
 /**
- * Initialise a virtual channel dechunker
+ * Returned from dyn_dechunker_process_data_chunk() and
+ * Returned from dyn_dechunker_process_first_chunk()
+ */
+enum dyn_dechunker_status
+{
+    E_DYN_INLINE_CHUNK = 0, ///< This chunk is complete in itself
+    E_DYN_IN_PROGRESS,      ///< The dechunker is processing chunks
+    E_DYN_READY,            ///< A dechunked stream is now complete
+    E_DYN_ERROR             ///< An error occurred (logged)
+};
+
+/**
+ * Initialise a static virtual channel dechunker
  *
  * @param chan_name - Name of channel
  * @param max_chunk_size - Max size of chunks allowed on channel
@@ -59,14 +72,14 @@ struct vc_dechunker *
 vc_dechunker_init(const char *chan_name, int max_chunk_size);
 
 /**
- * Free a virtual channel dechunker
+ * Free a static virtual channel dechunker
  * @param self vc dechunker to free
  */
 void
 vc_dechunker_free(struct vc_dechunker *self);
 
 /**
- * Process a virtual channel chunk
+ * Process a static virtual channel chunk
  *
  * @param self dechunker
  * @param s Stream for chunk, positioned at start of chunk
@@ -84,9 +97,9 @@ enum vc_dechunker_status
 vc_dechunker_process_chunk(struct vc_dechunker *self,
                            struct stream *s, int flags, int total_size);
 /**
- * Get the stream from a ready dechunker
+ * Get the stream from a ready static virtual dechunker
  *
- * @param self virtual channel dechunker
+ * @param self static virtual channel dechunker
  * @return input stream containing completed chunk
  *
  * Ownership of the stream passes to the caller
@@ -97,4 +110,80 @@ struct stream *
 vc_dechunker_get_stream(struct vc_dechunker *self);
 
 
+/**
+ * Initialise a dynamic virtual channel dechunker
+ *
+ * @param chan_name - Name of channel
+ * @return dyn_dechunker
+ */
+struct dyn_dechunker *
+dyn_dechunker_init(const char *chan_name);
+
+/**
+ * Free a dynamic channel dechunker
+ * @param self dynamic dechunker to free
+ */
+void
+dyn_dechunker_free(struct dyn_dechunker *self);
+
+/**
+ * Process a dynamic channel DYNVC_DATA_FIRST PDU ([MS-RDPEDYC] 2.2.3.1)
+ *
+ * @param self dechunker
+ * @param s Stream for chunk, positioned at start of data
+ * @param total_size length from DYNVC_DATA_FIRST header
+ * @return status of dechunker
+ *
+ * For PDUs of size > 1590 bytes, but < 1600, it is possible for the
+ * status E_DYN_INLINE_CHUNK to be returned, as the first chunk
+ * is complete in itself. This follows from [MS-RDPEDYC] 1.3.3.2.1 and
+ * 2.2.3.1
+ *
+ * E_DYN_READY will not be returned by this call.
+ *
+ * On error, it is not possible to recover the stream, because of the
+ * impossibility of distinguishing self-contained DATA PDUs, and
+ * DATA PDUs which are part of a larger PDU.
+ */
+enum dyn_dechunker_status
+dyn_dechunker_process_first_chunk(struct dyn_dechunker *self,
+                                  struct stream *s, int total_size);
+
+/**
+ * Process a dynamic channel DYNVC_DATA PDU ([MS-RDPEDYC] 2.2.3.2)
+ *
+ * @param self dechunker
+ * @param s Stream for chunk, positioned at start of data
+ * @return status of dechunker
+ *
+ * On error, it is not possible to recover the stream, because of the
+ * impossibility of distinguishing self-contained DATA PDUs, and
+ * DATA PDUs which are part of a larger PDU.
+ */
+enum dyn_dechunker_status
+dyn_dechunker_process_data_chunk(struct dyn_dechunker *self,
+                                 struct stream *s);
+
+/**
+ * Get the stream from a ready dynamic dechunker
+ *
+ * @param self dynamic channel dechunker
+ * @return input stream containing completed chunk
+ *
+ * Ownership of the stream passes to the caller
+ *
+ * Resets the dechunker state so that further chunks can be processed.
+ */
+struct stream *
+dyn_dechunker_get_stream(struct dyn_dechunker *self);
+
+/**
+ * Queries a dynamic dechunker for pending data
+ *
+ * @param self dynamic channel dechunker
+ * @return != 0 if data is stored in the dechunker
+ */
+int
+dyn_dechunker_pending(struct dyn_dechunker *self);
+
 #endif // DECHUNKER_H

common/xrdp_constants.h

@@ -87,6 +87,9 @@
 #define XRDP_MAX_BITMAP_CACHE_IDX 2000
 #define XRDP_BITMAP_CACHE_ENTRIES 2048
 
+/* Max number of dynamic channels we support */
+#define DRDYNVC_CHANNEL_COUNT 256
+
 /*
  * Constants come from ITU-T Recommendations
  */

libxrdp/libxrdp.h

@@ -143,12 +143,13 @@ struct xrdp_drdynvc
     int status; /* see XRDP_DRDYNVC_STATUS_* */
     int flags;
     int pad0;
+    struct dyn_dechunker *dc; // Use to dechunk fragments
     int (*open_response)(struct xrdp_process *id, int chan_id,
                          int creation_status);
     int (*close_response)(struct xrdp_process *id, int chan_id);
-    int (*data_first)(struct xrdp_process *id, int chan_id, char *data,
+    int (*data_first)(struct xrdp_process *id, int chan_id, struct stream *s,
-                      int bytes, int total_bytes);
+                      int total_bytes);
-    int (*data)(struct xrdp_process *id, int chan_id, char *data, int bytes);
+    int (*data)(struct xrdp_process *id, int chan_id, struct stream *s);
 };
 
 struct vc_dechunker; // Forward declaration
@@ -161,7 +162,7 @@ struct xrdp_channel
     int drdynvc_channel_id;
     int drdynvc_state;
     struct vc_dechunker *drdynvc_dc;
-    struct xrdp_drdynvc drdynvcs[256];
+    struct xrdp_drdynvc drdynvcs[DRDYNVC_CHANNEL_COUNT];
 };
 
 /* rdp */

libxrdp/libxrdpinc.h

@@ -91,9 +91,12 @@ struct xrdp_drdynvc_procs
     int (*open_response)(struct xrdp_process *id, int chan_id,
                          int creation_status);
     int (*close_response)(struct xrdp_process *id, int chan_id);
-    int (*data_first)(struct xrdp_process *id, int chan_id,
+    // Only set data_first if you want to be responsible for
-                      char *data, int bytes, int total_bytes);
+    // defragging your own PDUs. Otherwise the channel
-    int (*data)(struct xrdp_process *id, int chan_id, char *data, int bytes);
+    // process will do it for you.
+    int (*data_first)(struct xrdp_process *id, int chan_id, struct stream *s,
+                      int total_bytes);
+    int (*data)(struct xrdp_process *id, int chan_id, struct stream *s);
 };
 
 /* Defined in xrdp_client_info.h */

libxrdp/xrdp_channel.c

@@ -74,11 +74,16 @@ xrdp_channel_create(struct xrdp_sec *owner, struct xrdp_mcs *mcs_layer)
 void
 xrdp_channel_delete(struct xrdp_channel *self)
 {
+    int i;
     if (self == 0)
     {
         return;
     }
     vc_dechunker_free(self->drdynvc_dc);
+    for (i = 0 ; i < DRDYNVC_CHANNEL_COUNT ; ++i)
+    {
+        dyn_dechunker_free(self->drdynvcs[i].dc);
+    }
     g_memset(self, 0, sizeof(struct xrdp_channel));
     g_free(self);
 }
@@ -338,10 +343,10 @@ drdynvc_process_open_channel_response(struct xrdp_channel *self,
     in_uint32_le(s, creation_status); /* CreationStatus */
     LOG_DEVEL(LOG_LEVEL_TRACE, "Received [MS-RDPEDYC] DYNVC_CREATE_RSP "
               "ChannelId %d, CreationStatus %d", chan_id, creation_status);
-    if (chan_id > 255)
+    if (chan_id >= DRDYNVC_CHANNEL_COUNT)
     {
-        LOG(LOG_LEVEL_ERROR, "Received [MS-RDPEDYC] DYNVC_CREATE_RSP for an "
+        LOG(LOG_LEVEL_ERROR, "Received [MS-RDPEDYC] DYNVC_CREATE_RSP "
-            "invalid channel id. Max allowed 255, received %d", chan_id);
+            "for an invalid channel id %d", chan_id);
         return 1;
     }
 
@@ -354,6 +359,8 @@ drdynvc_process_open_channel_response(struct xrdp_channel *self,
     else
     {
         drdynvc->status = XRDP_DRDYNVC_STATUS_CLOSED;
+        dyn_dechunker_free(drdynvc->dc);
+        drdynvc->dc = NULL;
     }
     LOG_DEVEL(LOG_LEVEL_DEBUG,
               "Dynamic Virtual Channel %s (%d) updated: status = %s",
@@ -392,15 +399,28 @@ drdynvc_process_close_channel_response(struct xrdp_channel *self,
     LOG_DEVEL(LOG_LEVEL_TRACE, "Received [MS-RDPEDYC] DYNVC_CLOSE "
               "ChannelId %d", chan_id);
     session = self->sec_layer->rdp_layer->session;
-    if (chan_id > 255)
+    if (chan_id >= DRDYNVC_CHANNEL_COUNT)
     {
         LOG(LOG_LEVEL_ERROR, "Received message for an invalid "
-            "channel id. channel id %d", chan_id);
+            "channel id %d", chan_id);
         return 1;
     }
 
     drdynvc = self->drdynvcs + chan_id;
+
+    if (dyn_dechunker_pending(drdynvc->dc))
+    {
+        // The last PDU wasn't completed
+        LOG(LOG_LEVEL_WARNING,
+            "Dynamic Virtual Channel %s (%d) closing with outstanding data",
+            XRDP_DRDYNVC_CHANNEL_ID_TO_NAME(self, chan_id),
+            chan_id);
+    }
+
     drdynvc->status = XRDP_DRDYNVC_STATUS_CLOSED;
+    dyn_dechunker_free(drdynvc->dc);
+    drdynvc->dc = NULL;
+
     LOG_DEVEL(LOG_LEVEL_DEBUG,
               "Dynamic Virtual Channel %s (%d) updated: status = %s",
               XRDP_DRDYNVC_CHANNEL_ID_TO_NAME(self, chan_id),
@@ -426,76 +446,113 @@ static int
 drdynvc_process_data_first(struct xrdp_channel *self,
                            int cmd, struct stream *s)
 {
-    struct xrdp_session *session;
     uint32_t chan_id;
-    int len;
+    int rv = 0;
-    int bytes;
-    int total_bytes;
-    struct xrdp_drdynvc *drdynvc;
-
     if (drdynvc_get_chan_id(s, cmd, &chan_id) != 0) /* ChannelId */
     {
         LOG(LOG_LEVEL_ERROR,
-            "Parsing [MS-RDPEDYC] DYNVC_DATA_FIRST failed");
+            "Parsing [MS-RDPEDYC] DYNVC_DATA_FIRST channel ID failed");
-        return 1;
+        rv = 1;
-    }
-    len = (cmd >> 2) & 0x03;
-    if (len == 0)
-    {
-        if (!s_check_rem_and_log(s, 1, "Parsing [MS-RDPEDYC] DYNVC_DATA_FIRST"))
-        {
-            return 1;
-        }
-        in_uint8(s, total_bytes); /* Length */
-    }
-    else if (len == 1)
-    {
-        if (!s_check_rem_and_log(s, 2, "Parsing [MS-RDPEDYC] DYNVC_DATA_FIRST"))
-        {
-            return 1;
-        }
-        in_uint16_le(s, total_bytes); /* Length */
     }
     else
     {
-        if (!s_check_rem_and_log(s, 4, "Parsing [MS-RDPEDYC] DYNVC_DATA_FIRST"))
+        int len = (cmd >> 2) & 0x03;
+        int total_bytes;
+        switch (len)
         {
-            return 1;
+            case 0:
+                // Technically this can't happen as DATA_FIRST is only used for
+                // PDUs over 1590 bytes ([MS-RDPEDYC] 2.2.3)
+                if (!s_check_rem(s, 1))
+                {
+                    goto short_pdu;
+                }
+                in_uint8(s, total_bytes); /* Length */
+                break;
+
+            case 1:
+                if (!s_check_rem(s, 2))
+                {
+                    goto short_pdu;
+                }
+                in_uint16_le(s, total_bytes); /* Length */
+                break;
+
+            case 2:
+                if (!s_check_rem(s, 4))
+                {
+                    goto short_pdu;
                 }
                 in_uint32_le(s, total_bytes); /* Length */
+                break;
+
+            default:
+                LOG(LOG_LEVEL_ERROR,
+                    "[MS-RDPEDYC] DYNVC_DATA_FIRST has bad Len field");
+                return 1;
         }
-    bytes = (int) (s->end - s->p);
+
         LOG_DEVEL(LOG_LEVEL_TRACE, "Received [MS-RDPEDYC] DYNVC_DATA_FIRST "
                   "ChannelId %d, Length %d, Data (omitted from the log)",
                   chan_id, total_bytes);
 
-    // See [MS-RDPBCGR] 2.2.3
+        if (chan_id >= DRDYNVC_CHANNEL_COUNT)
-    if (total_bytes < 1590 || bytes > total_bytes)
         {
-        LOG(LOG_LEVEL_ERROR,
+            LOG(LOG_LEVEL_ERROR, "Received [MS-RDPEDYC] DYNVC_DATA_FIRST "
-            "Badly formed DYNVC_DATA_FIRST PDU received on dynamic channel %d",
+                "for an invalid channel id %d", chan_id);
-            chan_id);
+            rv = 1;
-        return 1;
         }
+        else
+        {
+            struct xrdp_drdynvc *drdynvc = self->drdynvcs + chan_id;
 
-    session = self->sec_layer->rdp_layer->session;
-    if (chan_id > 255)
-    {
-        LOG(LOG_LEVEL_ERROR, "Received [MS-RDPEDYC] DYNVC_DATA_FIRST for an "
-            "invalid channel id. Max allowed 255, received %d", chan_id);
-        return 1;
-    }
-    drdynvc = self->drdynvcs + chan_id;
             if (drdynvc->data_first != NULL)
             {
-        return drdynvc->data_first(session->id, chan_id, s->p,
+                // Caller has requested to defragment PDUs themself
-                                   bytes, total_bytes);
+                struct xrdp_session *session =
+                        self->sec_layer->rdp_layer->session;
+                rv = drdynvc->data_first(session->id, chan_id, s, total_bytes);
             }
-    LOG_DEVEL(LOG_LEVEL_WARNING, "Dynamic Virtual Channel %s (%d): "
+            else
-              "callback 'data_first' is NULL",
+            {
-              XRDP_DRDYNVC_CHANNEL_ID_TO_NAME(self, chan_id),
+                // Get the dechunker working on the stream
-              chan_id);
+                enum dyn_dechunker_status status;
-    return 0;
+                status = dyn_dechunker_process_first_chunk(drdynvc->dc,
+                         s, total_bytes);
+                switch (status)
+                {
+                    case E_DYN_INLINE_CHUNK:
+                        // Pass through as data PDU
+                        if (drdynvc->data != NULL)
+                        {
+                            struct xrdp_session *session =
+                                    self->sec_layer->rdp_layer->session;
+                            rv = drdynvc->data(session->id, chan_id, s);
+                        }
+                        break;
+
+                    case E_DYN_IN_PROGRESS:
+                        break;
+
+                    case E_DYN_ERROR:
+                        rv = 1;
+                        break;
+
+                    default:
+                        LOG(LOG_LEVEL_ERROR,
+                            "Dechunker returned unknown error %d",
+                            (int)status);
+                        rv = 1;
+                }
+            }
+        }
+    }
+
+    return rv;
+
+short_pdu:
+    LOG(LOG_LEVEL_ERROR, "[MS-RDPEDYC] DYNVC_DATA_FIRST is too short");
+    return 1;
 }
 
 /*****************************************************************************/
@@ -506,37 +563,86 @@ static int
 drdynvc_process_data(struct xrdp_channel *self,
                      int cmd, struct stream *s)
 {
-    struct xrdp_session *session;
+    int rv = 0;
     uint32_t chan_id;
-    int bytes;
-    struct xrdp_drdynvc *drdynvc;
 
     if (drdynvc_get_chan_id(s, cmd, &chan_id) != 0) /* ChannelId */
     {
-        LOG(LOG_LEVEL_ERROR, "drdynvc_process_data: drdynvc_get_chan_id failed");
+        LOG(LOG_LEVEL_ERROR,
-        return 1;
+            "Parsing [MS-RDPEDYC] DYNVC_DATA channel ID failed");
+        rv = 1;
     }
-    bytes = (int) (s->end - s->p);
+    else
+    {
         LOG_DEVEL(LOG_LEVEL_TRACE, "Received [MS-RDPEDYC] DYNVC_DATA "
-              "ChannelId %d, (re-assembled) Length %d, Data (omitted from the log)",
+                  "ChannelId %d, Length %d, Data (omitted from the log)",
-              chan_id, bytes);
+                  chan_id, s_rem(s));
-    session = self->sec_layer->rdp_layer->session;
+        if (chan_id >= DRDYNVC_CHANNEL_COUNT)
-    if (chan_id > 255)
         {
             LOG(LOG_LEVEL_ERROR, "Received DYNVC_DATA PDU for an invalid "
-            "channel id. channel id %d", chan_id);
+                "channel id %d", chan_id);
-        return 1;
+            rv = 1;
         }
-    drdynvc = self->drdynvcs + chan_id;
+        else
+        {
+            struct stream *ls = NULL; // Set if the application to be called
+            int free_ls = 0;    // Set if we need to clear ls when we're done
+            struct xrdp_drdynvc *drdynvc = self->drdynvcs + chan_id;
+            if (drdynvc->data_first != NULL)
+            {
+                // Caller is processing all PDUs directly
+                ls = s;
+            }
+            else
+            {
+                // Pass the PDU to the dechunker
+                enum dyn_dechunker_status dechunker_status =
+                    dyn_dechunker_process_data_chunk(drdynvc->dc, s);
+                switch (dechunker_status)
+                {
+                    case E_DYN_INLINE_CHUNK:
+                        ls = s;
+                        break;
+
+                    case E_DYN_IN_PROGRESS:
+                        break;
+
+                    case E_DYN_READY:
+                        ls = dyn_dechunker_get_stream(drdynvc->dc);
+                        // We now own the stream, so must delete it
+                        free_ls = 1;
+                        break;
+
+                    case E_DYN_ERROR:
+                        // Error has been logged
+                        rv = 1;
+                        break;
+
+                    default:
+                        LOG(LOG_LEVEL_ERROR,
+                            "Dechunker returned unknown error %d",
+                            (int)dechunker_status);
+                        rv = 1;
+                }
+            }
+
+            if (ls != NULL)
+            {
                 if (drdynvc->data != NULL)
                 {
-        return drdynvc->data(session->id, chan_id, s->p, bytes);
+                    struct xrdp_session *session =
+                            self->sec_layer->rdp_layer->session;
+                    rv = drdynvc->data(session->id, chan_id, ls);
                 }
-    LOG_DEVEL(LOG_LEVEL_WARNING, "Dynamic Virtual Channel %s (%d): "
+
-              "callback 'data' is NULL",
+                if (free_ls)
-              XRDP_DRDYNVC_CHANNEL_ID_TO_NAME(self, chan_id),
+                {
-              chan_id);
+                    free_stream(ls);
-    return 0;
+                }
+            }
+        }
+    }
+    return rv;
 }
 
 /*****************************************************************************/
@@ -820,6 +926,7 @@ xrdp_channel_drdynvc_open(struct xrdp_channel *self, const char *name,
     int total_data_len;
     int static_flags;
     char *cmd_ptr;
+    struct dyn_dechunker *dc = NULL;
 
     make_stream(s);
     init_stream(s, 8192);
@@ -827,8 +934,18 @@ xrdp_channel_drdynvc_open(struct xrdp_channel *self, const char *name,
     {
         LOG(LOG_LEVEL_ERROR,
             "xrdp_channel_drdynvc_open: xrdp_channel_init failed");
-        free_stream(s);
+        goto cleanup;
-        return 1;
+    }
+
+    // If there's no 'data_first' proc, we need a dechunker for the
+    // channel
+    if (procs->data_first == NULL)
+    {
+        if ((dc = dyn_dechunker_init(name)) == NULL)
+        {
+            // Error logged
+            goto cleanup;
+        }
     }
     cmd_ptr = s->p;
     out_uint8(s, 0); /* set later */
@@ -836,14 +953,12 @@ xrdp_channel_drdynvc_open(struct xrdp_channel *self, const char *name,
     while (self->drdynvcs[ChId].status != XRDP_DRDYNVC_STATUS_CLOSED)
     {
         ChId++;
-        if (ChId > 255)
+        if (ChId >= DRDYNVC_CHANNEL_COUNT)
         {
             LOG(LOG_LEVEL_ERROR,
                 "Attempting to create a new channel when the maximum "
-                "number of channels have already been created. "
+                "number of channels have already been created.");
-                "XRDP only supports 255 open channels.");
+            goto cleanup;
-            free_stream(s);
-            return 1;
         }
     }
     cbChId = drdynvc_insert_uint_124(s, ChId); /* ChannelId */
@@ -865,17 +980,30 @@ xrdp_channel_drdynvc_open(struct xrdp_channel *self, const char *name,
     {
         LOG(LOG_LEVEL_ERROR,
             "Sending [MS-RDPEDYC] DYNVC_CREATE_REQ failed");
-        free_stream(s);
+        goto cleanup;
-        return 1;
     }
-    free_stream(s);
+
+    if (procs->data == NULL)
+    {
+        // Not expected. Log this once.
+        LOG(LOG_LEVEL_WARNING, "Dynamic Virtual Channel %s (%d): "
+            "callback 'data' is NULL", name, ChId);
+    }
+
     *chan_id = ChId;
     self->drdynvcs[ChId].open_response = procs->open_response;
     self->drdynvcs[ChId].close_response = procs->close_response;
     self->drdynvcs[ChId].data_first = procs->data_first;
     self->drdynvcs[ChId].data = procs->data;
     self->drdynvcs[ChId].status = XRDP_DRDYNVC_STATUS_OPEN_SENT;
+    self->drdynvcs[ChId].dc = dc;
+    free_stream(s);
     return 0;
+
+cleanup:
+    free_stream(s);
+    dyn_dechunker_free(dc);
+    return 1;
 }
 
 /*****************************************************************************/
@@ -893,9 +1021,9 @@ xrdp_channel_drdynvc_close(struct xrdp_channel *self, int chan_id)
     int static_flags;
     char *cmd_ptr;
 
-    if ((chan_id < 0) || (chan_id > 255))
+    if ((chan_id < 0) || (chan_id >= DRDYNVC_CHANNEL_COUNT))
     {
-        LOG(LOG_LEVEL_ERROR, "Attempting to close an invalid channel id. "
+        LOG(LOG_LEVEL_ERROR, "Attempting to close an invalid "
             "channel id %d", chan_id);
         return 1;
     }
@@ -963,10 +1091,10 @@ xrdp_channel_drdynvc_data_first(struct xrdp_channel *self, int chan_id,
     int static_flags;
     char *cmd_ptr;
 
-    if ((chan_id < 0) || (chan_id > 255))
+    if ((chan_id < 0) || (chan_id >= DRDYNVC_CHANNEL_COUNT))
     {
         LOG(LOG_LEVEL_ERROR, "Attempting to send data to an invalid "
-            "channel id. channel id %d", chan_id);
+            "channel id %d", chan_id);
         return 1;
     }
     if (self->drdynvcs[chan_id].status != XRDP_DRDYNVC_STATUS_OPEN)
@@ -1035,10 +1163,10 @@ xrdp_channel_drdynvc_data(struct xrdp_channel *self, int chan_id,
     int static_flags;
     char *cmd_ptr;
 
-    if ((chan_id < 0) || (chan_id > 255))
+    if ((chan_id < 0) || (chan_id >= DRDYNVC_CHANNEL_COUNT))
     {
         LOG(LOG_LEVEL_ERROR, "Attempting to send data to an invalid "
-            "channel id. channel id %d", chan_id);
+            "channel id %d", chan_id);
         return 1;
     }
     if (self->drdynvcs[chan_id].status != XRDP_DRDYNVC_STATUS_OPEN)

sesman/chansrv/audin.c

@@ -77,7 +77,6 @@ static struct xr_wave_format_ex g_pcm_44100 =
 
 static struct chansrv_drdynvc_procs g_audin_info;
 static int g_audin_chanid;
-static struct stream *g_in_s;
 
 static struct xr_wave_format_ex *g_server_formats[] =
 {
@@ -422,70 +421,9 @@ audin_close_response(int chan_id)
     LOG_DEVEL(LOG_LEVEL_INFO, "audin_close_response:");
     g_audin_chanid = 0;
     cleanup_client_formats();
-    free_stream(g_in_s);
-    g_in_s = NULL;
     return 0;
 }
 
-/*****************************************************************************/
-static int
-audin_data_fragment(int chan_id, char *data, int bytes)
-{
-    int rv;
-
-    LOG_DEVEL(LOG_LEVEL_DEBUG, "audin_data_fragment:");
-    if (!s_check_rem(g_in_s, bytes))
-    {
-        LOG_DEVEL(LOG_LEVEL_ERROR, "audin_data_fragment: error bytes %d left %d",
-                  bytes, (int) (g_in_s->end - g_in_s->p));
-        return 1;
-    }
-    out_uint8a(g_in_s, data, bytes);
-    if (g_in_s->p == g_in_s->end)
-    {
-        g_in_s->p = g_in_s->data;
-        rv = audin_process_msg(chan_id, g_in_s);
-        free_stream(g_in_s);
-        g_in_s = NULL;
-        return rv;
-    }
-    return 0;
-}
-
-/*****************************************************************************/
-static int
-audin_data_first(int chan_id, char *data, int bytes, int total_bytes)
-{
-    LOG_DEVEL(LOG_LEVEL_DEBUG, "audin_data_first:");
-    if (g_in_s != NULL)
-    {
-        LOG_DEVEL(LOG_LEVEL_ERROR, "audin_data_first: warning g_in_s is not nil");
-        free_stream(g_in_s);
-    }
-    make_stream(g_in_s);
-    init_stream(g_in_s, total_bytes);
-    g_in_s->end = g_in_s->data + total_bytes;
-    return audin_data_fragment(chan_id, data, bytes);
-}
-
-/*****************************************************************************/
-static int
-audin_data(int chan_id, char *data, int bytes)
-{
-    struct stream ls;
-
-    LOG_DEVEL_HEXDUMP(LOG_LEVEL_TRACE, "audin_data:", data, bytes);
-    if (g_in_s == NULL)
-    {
-        g_memset(&ls, 0, sizeof(ls));
-        ls.data = data;
-        ls.p = ls.data;
-        ls.end = ls.p + bytes;
-        return audin_process_msg(chan_id, &ls);
-    }
-    return audin_data_fragment(chan_id, data, bytes);
-}
-
 /*****************************************************************************/
 int
 audin_init(void)
@@ -494,10 +432,9 @@ audin_init(void)
     g_memset(&g_audin_info, 0, sizeof(g_audin_info));
     g_audin_info.open_response = audin_open_response;
     g_audin_info.close_response = audin_close_response;
-    g_audin_info.data_first = audin_data_first;
+    g_audin_info.data_first = NULL;
-    g_audin_info.data = audin_data;
+    g_audin_info.data = audin_process_msg;
     g_audin_chanid = 0;
-    g_in_s = NULL;
     return 0;
 }
 

sesman/chansrv/chansrv.c

@@ -46,6 +46,7 @@
 #include "xrdp_constants.h"
 #include "audin.h"
 #include "channel_defs.h"
+#include "dechunker.h"
 
 #include "scp.h"
 #include "scp_sync.h"
@@ -82,7 +83,6 @@ tbus g_exec_mutex;
 tbus g_exec_sem;
 int g_exec_pid = 0;
 
-#define ARRAYSIZE(x) (sizeof(x)/sizeof(*(x)))
 /* max total channel bytes size */
 #define MAX_CHANNEL_BYTES (1 * 1024 * 1024 * 1024) /* 1 GB */
 #define MAX_CHANNEL_FRAG_BYTES 1600
@@ -98,14 +98,15 @@ struct chansrv_drdynvc
     int status; /* see CHANSRV_DRDYNVC_STATUS_* */
     int flags;
     int pad0;
+    struct dyn_dechunker *dc; // Use to dechunk fragments
     int (*open_response)(int chan_id, int creation_status);
     int (*close_response)(int chan_id);
-    int (*data_first)(int chan_id, char *data, int bytes, int total_bytes);
+    int (*data_first)(int chan_id, struct stream *s, int total_bytes);
-    int (*data)(int chan_id, char *data, int bytes);
+    int (*data)(int chan_id, struct stream *s);
     struct trans *xrdp_api_trans;
 };
 
-static struct chansrv_drdynvc g_drdynvcs[256];
+static struct chansrv_drdynvc g_drdynvcs[DRDYNVC_CHANNEL_COUNT];
 
 /* data in struct trans::callback_data */
 struct xrdp_api_data
@@ -576,7 +577,7 @@ static int
 process_message_drdynvc_open_response(struct stream *s)
 {
     struct chansrv_drdynvc *drdynvc;
-    int chan_id;
+    uint32_t chan_id;
     int creation_status;
 
     LOG_DEVEL(LOG_LEVEL_DEBUG, "process_message_drdynvc_open_response:");
@@ -586,7 +587,7 @@ process_message_drdynvc_open_response(struct stream *s)
     }
     in_uint32_le(s, chan_id);
     in_uint32_le(s, creation_status);
-    if ((chan_id < 0) || (chan_id > 255))
+    if (chan_id >= DRDYNVC_CHANNEL_COUNT)
     {
         return 1;
     }
@@ -603,6 +604,8 @@ process_message_drdynvc_open_response(struct stream *s)
     else
     {
         drdynvc->status = CHANSRV_DRDYNVC_STATUS_CLOSED;
+        dyn_dechunker_free(drdynvc->dc);
+        drdynvc->dc = NULL;
     }
     if (drdynvc->open_response != NULL)
     {
@@ -621,7 +624,7 @@ static int
 process_message_drdynvc_close_response(struct stream *s)
 {
     struct chansrv_drdynvc *drdynvc;
-    int chan_id;
+    uint32_t chan_id;
 
     LOG_DEVEL(LOG_LEVEL_DEBUG, "process_message_drdynvc_close_response:");
     if (!s_check_rem(s, 4))
@@ -629,7 +632,7 @@ process_message_drdynvc_close_response(struct stream *s)
         return 1;
     }
     in_uint32_le(s, chan_id);
-    if ((chan_id < 0) || (chan_id > 255))
+    if (chan_id >= DRDYNVC_CHANNEL_COUNT)
     {
         return 1;
     }
@@ -640,6 +643,8 @@ process_message_drdynvc_close_response(struct stream *s)
         return 0;
     }
     drdynvc->status = CHANSRV_DRDYNVC_STATUS_CLOSED;
+    dyn_dechunker_free(drdynvc->dc);
+    drdynvc->dc = NULL;
     if (drdynvc->close_response != NULL)
     {
         if (drdynvc->close_response(chan_id) != 0)
@@ -657,37 +662,58 @@ static int
 process_message_drdynvc_data_first(struct stream *s)
 {
     struct chansrv_drdynvc *drdynvc;
-    int chan_id;
+    uint32_t chan_id;
-    int bytes;
     int total_bytes;
-    char *data;
+    int rv = 0;
 
     LOG_DEVEL(LOG_LEVEL_DEBUG, "process_message_drdynvc_data_first:");
-    if (!s_check_rem(s, 12))
+    if (!s_check_rem(s, 8))
     {
         return 1;
     }
     in_uint32_le(s, chan_id);
-    in_uint32_le(s, bytes);
     in_uint32_le(s, total_bytes);
-    if (!s_check_rem(s, bytes))
+    if (chan_id >= DRDYNVC_CHANNEL_COUNT)
-    {
-        return 1;
-    }
-    in_uint8p(s, data, bytes);
-    if ((chan_id < 0) || (chan_id > 255))
     {
         return 1;
     }
     drdynvc = g_drdynvcs + chan_id;
     if (drdynvc->data_first != NULL)
     {
-        if (drdynvc->data_first(chan_id, data, bytes, total_bytes) != 0)
+        // Caller has requested to defragment PDUs themself
+        rv = drdynvc->data_first(chan_id, s, total_bytes);
+    }
+    else
     {
-            return 1;
+        // Get the dechunker working on the stream
+        enum dyn_dechunker_status status;
+        status = dyn_dechunker_process_first_chunk(drdynvc->dc,
+                 s, total_bytes);
+        switch (status)
+        {
+            case E_DYN_INLINE_CHUNK:
+                // Pass through as data PDU
+                if (drdynvc->data != NULL)
+                {
+                    rv = drdynvc->data(chan_id, s);
+                }
+                break;
+
+            case E_DYN_IN_PROGRESS:
+                break;
+
+            case E_DYN_ERROR:
+                rv = 1;
+                break;
+
+            default:
+                LOG(LOG_LEVEL_ERROR,
+                    "Dechunker returned unknown error %d",
+                    (int)status);
+                rv = 1;
         }
     }
-    return 0;
+    return rv;
 }
 
 /*****************************************************************************/
@@ -697,31 +723,73 @@ static int
 process_message_drdynvc_data(struct stream *s)
 {
     struct chansrv_drdynvc *drdynvc;
-    int chan_id;
+    uint32_t chan_id;
-    int bytes;
+    struct stream *ls = NULL; // Set if the application to be called
-    char *data;
+    int free_ls = 0;    // Set if we need to clear ls when we're done
+    int rv = 0;
 
     LOG_DEVEL(LOG_LEVEL_DEBUG, "process_message_drdynvc_data:");
-    if (!s_check_rem(s, 8))
+    if (!s_check_rem(s, 4))
     {
         return 1;
     }
     in_uint32_le(s, chan_id);
-    in_uint32_le(s, bytes);
+    if (chan_id >= DRDYNVC_CHANNEL_COUNT)
-    if (!s_check_rem(s, bytes))
     {
         return 1;
     }
-    in_uint8p(s, data, bytes);
     drdynvc = g_drdynvcs + chan_id;
+    if (drdynvc->data_first != NULL)
+    {
+        // Caller is processing all PDUs directly
+        ls = s;
+    }
+    else
+    {
+        // Pass the PDU to the dechunker
+        enum dyn_dechunker_status dechunker_status =
+            dyn_dechunker_process_data_chunk(drdynvc->dc, s);
+        switch (dechunker_status)
+        {
+            case E_DYN_INLINE_CHUNK:
+                ls = s;
+                break;
+
+            case E_DYN_IN_PROGRESS:
+                rv = 0;
+                break;
+
+            case E_DYN_READY:
+                ls = dyn_dechunker_get_stream(drdynvc->dc);
+                // We now own the stream, so must delete it
+                free_ls = 1;
+                break;
+
+            case E_DYN_ERROR:
+                // Error has been logged
+                rv = 1;
+                break;
+
+            default:
+                LOG(LOG_LEVEL_ERROR,
+                    "Dechunker returned unknown error %d",
+                    (int)dechunker_status);
+                rv = 1;
+        }
+    }
+
+    if (ls != NULL)
+    {
         if (drdynvc->data != NULL)
         {
-        if (drdynvc->data(chan_id, data, bytes) != 0)
+            rv = drdynvc->data(chan_id, ls);
+        }
+        if (free_ls)
         {
-            return 1;
+            free_stream(ls);
         }
     }
-    return 0;
+    return rv;
 }
 
 /*****************************************************************************/
@@ -734,12 +802,13 @@ chansrv_drdynvc_open(const char *name, int flags,
     int name_bytes;
     int lchan_id;
     int error;
+    struct dyn_dechunker *dc = NULL;
 
     lchan_id = 1;
     while (g_drdynvcs[lchan_id].status != CHANSRV_DRDYNVC_STATUS_CLOSED)
     {
         lchan_id++;
-        if (lchan_id > 255)
+        if (lchan_id >= DRDYNVC_CHANNEL_COUNT)
         {
             return 1;
         }
@@ -749,6 +818,18 @@ chansrv_drdynvc_open(const char *name, int flags,
     {
         return 1;
     }
+
+    // If there's no 'data_first' proc, we need a dechunker for the
+    // channel
+    if (procs->data_first == NULL)
+    {
+        if ((dc = dyn_dechunker_init(name)) == NULL)
+        {
+            // Error logged
+            return 1;
+        }
+    }
+
     name_bytes = g_strlen(name);
     out_uint32_le(s, 0); /* version */
     out_uint32_le(s, 8 + 8 + 4 + name_bytes + 4 + 4);
@@ -765,13 +846,17 @@ chansrv_drdynvc_open(const char *name, int flags,
         if (chan_id != NULL)
         {
             *chan_id = lchan_id;
+        }
         g_drdynvcs[lchan_id].open_response = procs->open_response;
         g_drdynvcs[lchan_id].close_response = procs->close_response;
         g_drdynvcs[lchan_id].data_first = procs->data_first;
         g_drdynvcs[lchan_id].data = procs->data;
         g_drdynvcs[lchan_id].status = CHANSRV_DRDYNVC_STATUS_OPEN_SENT;
-
+        g_drdynvcs[lchan_id].dc = dc;
     }
+    else
+    {
+        dyn_dechunker_free(dc);
     }
     return error;
 }
@@ -804,6 +889,10 @@ chansrv_drdynvc_close(int chan_id)
     struct stream *s;
     int error;
 
+    if (chan_id < 0 || chan_id >= DRDYNVC_CHANNEL_COUNT)
+    {
+        return 1;
+    }
     s = trans_get_out_s(g_con_trans, 8192);
     if (s == NULL)
     {
@@ -1093,9 +1182,10 @@ my_trans_data_in(struct trans *trans)
 
 /*****************************************************************************/
 static struct trans *
-get_api_trans_from_chan_id(int chan_id)
+get_api_trans_from_chan_id(uint32_t chan_id)
 {
-    return g_drdynvcs[chan_id].xrdp_api_trans;
+    return (chan_id >= DRDYNVC_CHANNEL_COUNT)
+           ? NULL : g_drdynvcs[chan_id].xrdp_api_trans;
 }
 
 /*****************************************************************************/
@@ -1136,24 +1226,24 @@ my_api_close_response(int chan_id)
 
 /*****************************************************************************/
 static int
-my_api_data_first(int chan_id, char *data, int bytes, int total_bytes)
+my_api_data_first(int chan_id, struct stream *s, int total_bytes)
 {
     struct trans *trans;
-    struct stream *s;
+    struct stream *out_s;
-
+    int bytes = s_rem(s);
     //g_writeln("my_api_data_first: bytes %d total_bytes %d", bytes, total_bytes);
     trans = get_api_trans_from_chan_id(chan_id);
     if (trans == NULL)
     {
         return 1;
     }
-    s = trans_get_out_s(trans, bytes);
+    out_s = trans_get_out_s(trans, bytes);
-    if (s == NULL)
+    if (out_s == NULL)
     {
         return 1;
     }
-    out_uint8a(s, data, bytes);
+    out_uint8a(out_s, s->p, bytes);
-    s_mark_end(s);
+    s_mark_end(out_s);
     if (trans_write_copy(trans) != 0)
     {
         return 1;
@@ -1163,24 +1253,24 @@ my_api_data_first(int chan_id, char *data, int bytes, int total_bytes)
 
 /*****************************************************************************/
 static int
-my_api_data(int chan_id, char *data, int bytes)
+my_api_data(int chan_id, struct stream *s)
 {
     struct trans *trans;
-    struct stream *s;
+    struct stream *out_s;
+    int bytes = s_rem(s);
 
-    //g_writeln("my_api_data: bytes %d", bytes);
     trans = get_api_trans_from_chan_id(chan_id);
     if (trans == NULL)
     {
         return 1;
     }
-    s = trans_get_out_s(trans, bytes);
+    out_s = trans_get_out_s(trans, bytes);
-    if (s == NULL)
+    if (out_s == NULL)
     {
         return 1;
     }
-    out_uint8a(s, data, bytes);
+    out_uint8a(out_s, s->p, bytes);
-    s_mark_end(s);
+    s_mark_end(out_s);
     if (trans_write_copy(trans) != 0)
     {
         return 1;
@@ -1519,7 +1609,7 @@ api_con_trans_list_check_wait_objs(void)
                     chansrv_drdynvc_close(ad->chan_id);
                 }
                 for (drdynvc_index = 0;
-                        drdynvc_index < (int) ARRAYSIZE(g_drdynvcs);
+                        drdynvc_index < DRDYNVC_CHANNEL_COUNT;
                         drdynvc_index++)
                 {
                     if (g_drdynvcs[drdynvc_index].xrdp_api_trans == ltran)
@@ -1754,6 +1844,8 @@ x_server_fatal_handler(void)
 int
 main_cleanup(void)
 {
+    int i;
+
     if (g_term_event != 0)
     {
         g_delete_wait_obj(g_term_event);
@@ -1772,6 +1864,10 @@ main_cleanup(void)
         tc_mutex_delete(g_exec_mutex);
         tc_sem_delete(g_exec_sem);
     }
+    for (i = 0 ; i < DRDYNVC_CHANNEL_COUNT; ++i)
+    {
+        dyn_dechunker_free(g_drdynvcs[i].dc);
+    }
     log_end();
     config_free(g_cfg);
     g_deinit(); /* os_calls */

sesman/chansrv/chansrv.h

@@ -62,8 +62,10 @@ struct chansrv_drdynvc_procs
 {
     int (*open_response)(int chan_id, int creation_status);
     int (*close_response)(int chan_id);
-    int (*data_first)(int chan_id, char *data, int bytes, int total_bytes);
+    // Set data_first to NULL to have the dechunker automatically
-    int (*data)(int chan_id, char *data, int bytes);
+    // handle channel fragments
+    int (*data_first)(int chan_id, struct stream *s, int total_bytes);
+    int (*data)(int chan_id, struct stream *s);
 };
 
 int

tests/common/test_dechunker.c

@@ -143,13 +143,24 @@ static const char frankenstein[] =
     "all your love and kindness.\n\n"
     "Your affectionate brother,\nR. Walton ";
 
-// Number of CHANNEL_CHUNK_LENGTH chunks required to send the
+// Number of CHANNEL_CHUNK_LENGTH (1600) chunks required to send the
-// above text into the dechunker
+// above text into the vc dechunker
-#define FRANKENSTEIN_CHUNK_COUNT \
+#define FRANKENSTEIN_VC_CHUNK_COUNT \
-    ((sizeof(frankenstein) + (CHANNEL_CHUNK_LENGTH -1)) \
+    ((sizeof(frankenstein) + (CHANNEL_CHUNK_LENGTH - 1)) \
      / CHANNEL_CHUNK_LENGTH)
 
-// See the private E_MAX_CHUNK_SIZE_LOWER_LIMIT in dechunker.c
+// The dynamic dechunker works on total data block sizes of 1600 bytes,
+// including the block header as well.
+// The FIRST block header is 6-12 bytes long, and the DATA block header
+// is 5-8 bytes long. For simplicity we assume a header size of 8
+// bytes, and hence a data size of 1592 bytes.
+#define FRANKENSTEIN_DYN_CHUNK_SIZE 1592
+
+#define FRANKENSTEIN_DYN_CHUNK_COUNT \
+    ((sizeof(frankenstein) + (FRANKENSTEIN_DYN_CHUNK_SIZE - 1)) \
+     / FRANKENSTEIN_DYN_CHUNK_SIZE)
+
+// See the private E_MAX_VC_CHUNK_SIZE_LOWER_LIMIT in dechunker.c
 #define PAD50 "                                                  "
 
 /******************************************************************************/
@@ -175,7 +186,7 @@ make_stream_from_data(const char *data, int data_len)
 /*
  * Check bad parameters passed to the dechunker functions
  */
-START_TEST(test_dechunker_bad_params)
+START_TEST(test_vc_dechunker_bad_params)
 {
     struct vc_dechunker *dc;
     const char data[] = "Some stream data";
@@ -191,10 +202,10 @@ START_TEST(test_dechunker_bad_params)
     ck_assert_ptr_ne(dc, NULL);
 
     // vc_dechunker_free
-    vc_dechunker_free(NULL);   // Musn't crash!
+    vc_dechunker_free(NULL);   // Must not crash!
 
     // vc_dechunker_get_stream
-    vc_dechunker_get_stream(NULL);   // Musn't crash!
+    vc_dechunker_get_stream(NULL);   // Must not crash!
 
     // vc_dechunker_process_chunk
     stat = vc_dechunker_process_chunk(NULL, s, 0, 1600); // No dechunker
@@ -216,7 +227,7 @@ START_TEST(test_dechunker_bad_params)
  * immediately returned to the caller with E_VC_INLINE_CHUNK. If
  * however, we are currently dechunking, a passthrough chunk is not allowed
  */
-START_TEST(test_dechunker_passthrough)
+START_TEST(test_vc_dechunker_passthrough)
 {
     const char data[] = "Some data to dechunk";
     struct stream *s = make_stream_from_data(data, sizeof(data));
@@ -272,7 +283,7 @@ END_TEST
  * An intermediate chunk (neither first of last) must be rejected if we
  * are not dechunking
  */
-START_TEST(test_dechunker_intermediate)
+START_TEST(test_vc_dechunker_intermediate)
 {
     const char data[] = PAD50 "Some data to dechunk";
     struct stream *s = make_stream_from_data(data, sizeof(data));
@@ -297,7 +308,7 @@ END_TEST
 /*
  * A LAST chunk must be rejected if we are not dechunking
  */
-START_TEST(test_dechunker_last)
+START_TEST(test_vc_dechunker_last)
 {
     const char data[] = PAD50 "Some data to dechunk";
     struct stream *s = make_stream_from_data(data, sizeof(data));
@@ -322,7 +333,7 @@ END_TEST
 /**
  * Two consecutive FIRST chunks are not allowed
  */
-START_TEST(test_dechunker_first_first)
+START_TEST(test_vc_dechunker_first_first)
 {
     const char data[] = PAD50 "Some data to dechunk";
     struct stream *s = make_stream_from_data(data, sizeof(data));
@@ -367,7 +378,7 @@ END_TEST
 *  > Virtual channel data that fits in a single Virtual Channel PDU MUST
 *  > specify both flags
  */
-START_TEST(test_dechunker_chunk_overflow)
+START_TEST(test_vc_dechunker_chunk_overflow)
 {
     const char data[] = PAD50 "Some data to dechunk";
     struct stream *s = make_stream_from_data(data, sizeof(data));
@@ -408,15 +419,15 @@ END_TEST
 // up to CHANNEL_CHUNK_LENGTH bytes from Frankenstein chapter 1
 // The stream pointer will be positioned at the start of the text.
 static struct stream *
-make_bigtest_chunk(unsigned int chunk_num, int *flags)
+make_vc_bigtest_chunk(unsigned int chunk_num, int *flags)
 {
     struct stream *s = NULL;
 
-    if (chunk_num < FRANKENSTEIN_CHUNK_COUNT)
+    if (chunk_num < FRANKENSTEIN_VC_CHUNK_COUNT)
     {
         int chunk_size;
         // Work out the size of this chunk
-        if (chunk_num == (FRANKENSTEIN_CHUNK_COUNT - 1))
+        if (chunk_num == (FRANKENSTEIN_VC_CHUNK_COUNT - 1))
         {
             chunk_size = sizeof(frankenstein) % CHANNEL_CHUNK_LENGTH;
             if (chunk_size == 0)
@@ -456,7 +467,7 @@ make_bigtest_chunk(unsigned int chunk_num, int *flags)
         // Sort out the flags
         *flags =
             (chunk_num == 0) ? XR_CHANNEL_FLAG_FIRST :
-            (chunk_num == (FRANKENSTEIN_CHUNK_COUNT - 1)) ? XR_CHANNEL_FLAG_LAST :
+            (chunk_num == (FRANKENSTEIN_VC_CHUNK_COUNT - 1)) ? XR_CHANNEL_FLAG_LAST :
             0;
     }
 
@@ -468,7 +479,7 @@ make_bigtest_chunk(unsigned int chunk_num, int *flags)
  * Streams a lot of data through the dechunker and checks it's all
  * assembled correctly at the end
  */
-START_TEST(test_dechunker_big_test)
+START_TEST(test_vc_dechunker_big_test)
 {
     enum vc_dechunker_status stat;
     struct vc_dechunker *dc = vc_dechunker_init("test", CHANNEL_CHUNK_LENGTH);
@@ -476,15 +487,15 @@ START_TEST(test_dechunker_big_test)
     struct stream *s;
 
     int i;
-    for (i = 0 ; i < FRANKENSTEIN_CHUNK_COUNT; ++i)
+    for (i = 0 ; i < FRANKENSTEIN_VC_CHUNK_COUNT; ++i)
     {
         int flags;
-        s = make_bigtest_chunk(i, &flags);
+        s = make_vc_bigtest_chunk(i, &flags);
         stat = vc_dechunker_process_chunk(
                    dc, s,
                    flags,
                    sizeof(frankenstein));
-        if (i < FRANKENSTEIN_CHUNK_COUNT - 1)
+        if (i < FRANKENSTEIN_VC_CHUNK_COUNT - 1)
         {
             ck_assert_int_eq(stat, E_VC_IN_PROGRESS);
         }
@@ -513,8 +524,8 @@ START_TEST(test_dechunker_big_test)
 }
 
 /******************************************************************************/
-// Like test_dechunker_big_test, but the last chunk is oversized
+// Like test_vc_dechunker_big_test, but the last chunk is oversized
-START_TEST(test_dechunker_big_test_oversize_fail)
+START_TEST(test_vc_dechunker_big_test_oversize_fail)
 {
     enum vc_dechunker_status stat;
     struct vc_dechunker *dc = vc_dechunker_init("test", CHANNEL_CHUNK_LENGTH);
@@ -522,11 +533,11 @@ START_TEST(test_dechunker_big_test_oversize_fail)
     struct stream *s;
 
     int i;
-    for (i = 0 ; i < FRANKENSTEIN_CHUNK_COUNT; ++i)
+    for (i = 0 ; i < FRANKENSTEIN_VC_CHUNK_COUNT; ++i)
     {
         int flags;
-        s = make_bigtest_chunk(i, &flags);
+        s = make_vc_bigtest_chunk(i, &flags);
-        if (i == (FRANKENSTEIN_CHUNK_COUNT - 1))
+        if (i == (FRANKENSTEIN_VC_CHUNK_COUNT - 1))
         {
             // Add a byte to the end of the text in the chunk
             struct stream *s2;
@@ -546,7 +557,7 @@ START_TEST(test_dechunker_big_test_oversize_fail)
                    dc, s,
                    flags,
                    sizeof(frankenstein));
-        if (i < FRANKENSTEIN_CHUNK_COUNT - 1)
+        if (i < FRANKENSTEIN_VC_CHUNK_COUNT - 1)
         {
             ck_assert_int_eq(stat, E_VC_IN_PROGRESS);
         }
@@ -561,8 +572,8 @@ START_TEST(test_dechunker_big_test_oversize_fail)
 }
 
 /******************************************************************************/
-// Like test_dechunker_big_test, but the last chunk is undersized
+// Like test_vc_dechunker_big_test, but the last chunk is undersized
-START_TEST(test_dechunker_big_test_undersize_fail)
+START_TEST(test_vc_dechunker_big_test_undersize_fail)
 {
     enum vc_dechunker_status stat;
     struct vc_dechunker *dc = vc_dechunker_init("test", CHANNEL_CHUNK_LENGTH);
@@ -570,11 +581,11 @@ START_TEST(test_dechunker_big_test_undersize_fail)
     struct stream *s;
 
     int i;
-    for (i = 0 ; i < FRANKENSTEIN_CHUNK_COUNT; ++i)
+    for (i = 0 ; i < FRANKENSTEIN_VC_CHUNK_COUNT; ++i)
     {
         int flags;
-        s = make_bigtest_chunk(i, &flags);
+        s = make_vc_bigtest_chunk(i, &flags);
-        if (i == (FRANKENSTEIN_CHUNK_COUNT - 1))
+        if (i == (FRANKENSTEIN_VC_CHUNK_COUNT - 1))
         {
             // Skip a byte in the stream, so there is one
             // less byte than expected
@@ -584,7 +595,7 @@ START_TEST(test_dechunker_big_test_undersize_fail)
                    dc, s,
                    flags,
                    sizeof(frankenstein));
-        if (i < FRANKENSTEIN_CHUNK_COUNT - 1)
+        if (i < FRANKENSTEIN_VC_CHUNK_COUNT - 1)
         {
             ck_assert_int_eq(stat, E_VC_IN_PROGRESS);
         }
@@ -600,6 +611,344 @@ START_TEST(test_dechunker_big_test_undersize_fail)
 
 /******************************************************************************/
 
+/******************************************************************************/
+/*
+ * Check bad parameters passed to the dechunker functions
+ */
+START_TEST(test_dyn_dechunker_bad_params)
+{
+    struct dyn_dechunker *dc;
+    const char data[] = "Some stream data";
+    struct stream *s = make_stream_from_data(data, sizeof(data));
+    enum dyn_dechunker_status stat;
+
+    // dyn_dechunker_init
+    dc = dyn_dechunker_init(NULL); // No channel name
+    ck_assert_ptr_eq(dc, NULL);
+    dc = dyn_dechunker_init("test"); // Should be OK
+    ck_assert_ptr_ne(dc, NULL);
+
+    // dyn_dechunker_free
+    dyn_dechunker_free(NULL);   // Must not crash!
+
+    // dyn_dechunker_get_stream
+    dyn_dechunker_get_stream(NULL);   // Must not crash!
+
+    // dyn_dechunker_process_first_chunk
+    stat = dyn_dechunker_process_first_chunk(NULL, s, 1600); // No dechunker
+    ck_assert_int_eq(stat, E_DYN_ERROR);
+    stat = dyn_dechunker_process_first_chunk(dc, NULL, 1600); // No stream
+    ck_assert_int_eq(stat, E_DYN_ERROR);
+    stat = dyn_dechunker_process_first_chunk(dc, s, -1); // bad total_size
+    ck_assert_int_eq(stat, E_DYN_ERROR);
+
+    // dyn_dechunker_process_data_chunk
+    stat = dyn_dechunker_process_data_chunk(NULL, s); // No dechunker
+    ck_assert_int_eq(stat, E_DYN_ERROR);
+    stat = dyn_dechunker_process_data_chunk(dc, NULL); // No stream
+    ck_assert_int_eq(stat, E_DYN_ERROR);
+
+    free_stream(s);
+    dyn_dechunker_free(dc);
+}
+
+/******************************************************************************/
+/*
+ * Check passthrough DATA chunks (i.e. those not following a FIRST)
+ *
+ * When the dechunker is in normal operation, these chunks are
+ * immediately returned to the caller with E_DYN_INLINE_CHUNK.
+ */
+START_TEST(test_dyn_dechunker_passthrough)
+{
+    const char data[] = "Some data to dechunk";
+    struct stream *s = make_stream_from_data(data, sizeof(data));
+    enum dyn_dechunker_status stat;
+
+    struct dyn_dechunker *dc = dyn_dechunker_init("test");
+    ck_assert_ptr_ne(dc, NULL);
+
+    // Save the stream pointer so we can reset the stream in between calls
+    s_push_layer(s, iso_hdr, 0);
+
+    // Check a DATA chunk is normally recognised immediately
+    stat = dyn_dechunker_process_data_chunk(
+               dc, s);
+    ck_assert_int_eq(stat, E_DYN_INLINE_CHUNK);
+
+    dyn_dechunker_free(dc);
+    free_stream(s);
+}
+END_TEST
+
+
+/******************************************************************************/
+/**
+ * Two consecutive FIRST chunks are not allowed
+ */
+START_TEST(test_dyn_dechunker_first_first)
+{
+    const char data[] = PAD50 "Some data to dechunk";
+    struct stream *s = make_stream_from_data(data, sizeof(data));
+    enum dyn_dechunker_status stat;
+
+    struct dyn_dechunker *dc = dyn_dechunker_init("test");
+    ck_assert_ptr_ne(dc, NULL);
+
+    // Save the stream pointer so we can reset the stream in between calls
+    s_push_layer(s, iso_hdr, 0);
+
+    // Check a FIRST chunk is accepted...
+    stat = dyn_dechunker_process_first_chunk(
+               dc, s, 1600);
+    ck_assert_int_eq(stat, E_DYN_IN_PROGRESS);
+
+    // ... and another FIRST chunk is an error
+    s_pop_layer(s, iso_hdr);
+    stat = dyn_dechunker_process_first_chunk(
+               dc, s, 1600);
+    ck_assert_int_eq(stat, E_DYN_ERROR);
+
+    dyn_dechunker_free(dc);
+    free_stream(s);
+}
+END_TEST
+
+
+/******************************************************************************/
+/**
+ * A FIRST chunk bigger than 1590 bytes but less than 1600 is passed
+ * through to the application, if it is the total length
+ */
+START_TEST(test_dyn_dechunker_first_inline)
+{
+    const char data[1591] = {0};
+
+    enum dyn_dechunker_status stat;
+    struct dyn_dechunker *dc;
+    struct stream *s;
+
+    // Check a FIRST chunk of 1590 bytes with a total of 1590 is rejected
+    // (too small to fragment)
+    dc = dyn_dechunker_init("test");
+    ck_assert_ptr_ne(dc, NULL);
+    s = make_stream_from_data(data, 1590);
+    ck_assert_ptr_ne(s, NULL);
+    stat = dyn_dechunker_process_first_chunk( dc, s, 1590);
+    ck_assert_int_eq(stat, E_DYN_ERROR);
+    dyn_dechunker_free(dc);
+    free_stream(s);
+
+    // Check a FIRST chunk of 1591 bytes with a total size of 1591 is
+    // accepted as inline
+    dc = dyn_dechunker_init("test");
+    ck_assert_ptr_ne(dc, NULL);
+    s = make_stream_from_data(data, 1591);
+    ck_assert_ptr_ne(s, NULL);
+    stat = dyn_dechunker_process_first_chunk( dc, s, 1591);
+    ck_assert_int_eq(stat, E_DYN_INLINE_CHUNK);
+    dyn_dechunker_free(dc);
+    free_stream(s);
+}
+END_TEST
+
+
+/******************************************************************************/
+/**
+ * Checks that a FIRST chunk cannot be bigger than the total size
+ */
+START_TEST(test_dyn_dechunker_chunk_overflow)
+{
+    const char data[1592] = {0};
+
+    enum dyn_dechunker_status stat;
+    struct dyn_dechunker *dc;
+    struct stream *s;
+
+    // We know a FIRST chunk of size 1591 for a total of 1591 is
+    // inline (see test_dyn_dechunker_first_inline()). Check if the
+    // first chunk is 1592, it is rejected
+    dc = dyn_dechunker_init("test");
+    ck_assert_ptr_ne(dc, NULL);
+    s = make_stream_from_data(data, 1592);
+    ck_assert_ptr_ne(s, NULL);
+    stat = dyn_dechunker_process_first_chunk( dc, s, 1591);
+    ck_assert_int_eq(stat, E_DYN_ERROR);
+    dyn_dechunker_free(dc);
+    free_stream(s);
+}
+END_TEST
+
+/******************************************************************************/
+// Returns a stream with some random data, then a chunk of
+// up to FRANKENSTEIN_DYN_CHUNK_COUNT bytes from Frankenstein chapter 1
+// The stream pointer will be positioned at the start of the text.
+static struct stream *
+make_dyn_bigtest_chunk(unsigned int chunk_num)
+{
+    struct stream *s = NULL;
+
+    if (chunk_num < FRANKENSTEIN_DYN_CHUNK_COUNT)
+    {
+        int chunk_size;
+        // Work out the size of this chunk
+        if (chunk_num == (FRANKENSTEIN_DYN_CHUNK_COUNT - 1))
+        {
+            chunk_size = sizeof(frankenstein) % FRANKENSTEIN_DYN_CHUNK_SIZE;
+            if (chunk_size == 0)
+            {
+                chunk_size = FRANKENSTEIN_DYN_CHUNK_SIZE;
+            }
+        }
+        else
+        {
+            chunk_size = FRANKENSTEIN_DYN_CHUNK_SIZE;
+        }
+
+        // Add some random data at the start of the stream, so we
+        // can check the dechunker works for non-zero positioned
+        // streams
+        int rand_size = 4 * 4 * chunk_num;
+
+        make_stream(s);
+        init_stream(s, rand_size + chunk_size);
+
+        // Write the random data
+        int i;
+        for (i = 0 ; i < rand_size; ++i)
+        {
+            out_uint8(s, rand() & 255);
+        }
+        s_push_layer(s, iso_hdr, 0);
+
+        // Copy the chapter data
+        out_uint8a(s, &frankenstein[chunk_num * FRANKENSTEIN_DYN_CHUNK_SIZE],
+                   chunk_size);
+
+        // Get the stream ready for reading from the Frankenstein text
+        s_mark_end(s);
+        s_pop_layer(s, iso_hdr);
+    }
+
+    return s;
+}
+
+/******************************************************************************/
+/*
+ * Streams a lot of data through the dechunker and checks it's all
+ * assembled correctly at the end
+ */
+START_TEST(test_dyn_dechunker_big_test)
+{
+    enum dyn_dechunker_status stat;
+    struct dyn_dechunker *dc = dyn_dechunker_init("test");
+    ck_assert_ptr_ne(dc, NULL);
+    struct stream *s;
+    int pending;
+
+    int i;
+    for (i = 0 ; i < FRANKENSTEIN_DYN_CHUNK_COUNT; ++i)
+    {
+        s = make_dyn_bigtest_chunk(i);
+        if (i == 0)
+        {
+            stat = dyn_dechunker_process_first_chunk(
+                       dc, s, sizeof(frankenstein));
+        }
+        else
+        {
+            stat = dyn_dechunker_process_data_chunk( dc, s);
+        }
+        pending = dyn_dechunker_pending(dc);
+        ck_assert_int_ne(pending, 0);
+        if (i < FRANKENSTEIN_DYN_CHUNK_COUNT - 1)
+        {
+            ck_assert_int_eq(stat, E_DYN_IN_PROGRESS);
+        }
+        else
+        {
+            ck_assert_int_eq(stat, E_DYN_READY);
+        }
+        free_stream(s);
+    }
+
+    // Check we have a result
+    s = dyn_dechunker_get_stream(dc);
+    ck_assert_ptr_ne(s, NULL);
+    pending = dyn_dechunker_pending(dc);
+    ck_assert_int_eq(pending, 0);
+
+    // Is it the right size?
+    int stream_size = s_rem(s);
+    ck_assert_int_eq(stream_size, sizeof(frankenstein));
+
+    // Check the data
+    const char *p;
+    in_uint8p(s, p, stream_size);
+    ck_assert_mem_eq(frankenstein, p, stream_size);
+
+    free_stream(s);
+    dyn_dechunker_free(dc);
+}
+
+/******************************************************************************/
+// Like test_dyn_dechunker_big_test, but the last chunk is oversized
+START_TEST(test_dyn_dechunker_big_test_oversize_fail)
+{
+    enum dyn_dechunker_status stat;
+    struct dyn_dechunker *dc = dyn_dechunker_init("test");
+    ck_assert_ptr_ne(dc, NULL);
+    struct stream *s;
+
+    int i;
+    for (i = 0 ; i < FRANKENSTEIN_DYN_CHUNK_COUNT; ++i)
+    {
+        s = make_dyn_bigtest_chunk(i);
+        if (i == 0)
+        {
+            stat = dyn_dechunker_process_first_chunk(
+                       dc, s, sizeof(frankenstein));
+        }
+        else
+        {
+            if (i == (FRANKENSTEIN_DYN_CHUNK_COUNT - 1))
+            {
+                // Add a byte to the end of the text in the chunk
+                struct stream *s2;
+                make_stream(s2);
+                init_stream(s2, s_rem(s) + 1);
+                s_push_layer(s2, iso_hdr, 0);
+                out_uint8p(s2, s->p, s_rem(s));
+                out_uint8(s2, 'x');
+                s_mark_end(s2);
+                s_pop_layer(s2, iso_hdr); // Rewind for reading
+                // Swap s2 and s and delete the original stream
+                struct stream *tmp = s;
+                s = s2;
+                free_stream(tmp);
+            }
+            stat = dyn_dechunker_process_data_chunk( dc, s);
+        }
+        if (i < FRANKENSTEIN_DYN_CHUNK_COUNT - 1)
+        {
+            ck_assert_int_eq(stat, E_DYN_IN_PROGRESS);
+        }
+        else
+        {
+            ck_assert_int_eq(stat, E_DYN_ERROR);
+        }
+        free_stream(s);
+    }
+
+    // Check we do not have a result
+    s = dyn_dechunker_get_stream(dc);
+    ck_assert_ptr_eq(s, NULL);
+
+    dyn_dechunker_free(dc);
+}
+
+/******************************************************************************/
+
 Suite *
 make_suite_test_dechunker(void)
 {
@@ -608,17 +957,27 @@ make_suite_test_dechunker(void)
 
     s = suite_create("dechunker");
 
-    rc_dechunker = tcase_create("dechunker_basic");
+    rc_dechunker = tcase_create("vc_dechunker");
     suite_add_tcase(s, rc_dechunker);
-    tcase_add_test(rc_dechunker, test_dechunker_bad_params);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_bad_params);
-    tcase_add_test(rc_dechunker, test_dechunker_passthrough);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_passthrough);
-    tcase_add_test(rc_dechunker, test_dechunker_intermediate);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_intermediate);
-    tcase_add_test(rc_dechunker, test_dechunker_last);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_last);
-    tcase_add_test(rc_dechunker, test_dechunker_first_first);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_first_first);
-    tcase_add_test(rc_dechunker, test_dechunker_chunk_overflow);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_chunk_overflow);
-    tcase_add_test(rc_dechunker, test_dechunker_big_test);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_big_test);
-    tcase_add_test(rc_dechunker, test_dechunker_big_test_oversize_fail);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_big_test_oversize_fail);
-    tcase_add_test(rc_dechunker, test_dechunker_big_test_undersize_fail);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_big_test_undersize_fail);
+
+    rc_dechunker = tcase_create("dyn_dechunker");
+    suite_add_tcase(s, rc_dechunker);
+    tcase_add_test(rc_dechunker, test_dyn_dechunker_bad_params);
+    tcase_add_test(rc_dechunker, test_dyn_dechunker_passthrough);
+    tcase_add_test(rc_dechunker, test_dyn_dechunker_first_first);
+    tcase_add_test(rc_dechunker, test_dyn_dechunker_first_inline);
+    tcase_add_test(rc_dechunker, test_dyn_dechunker_chunk_overflow);
+    tcase_add_test(rc_dechunker, test_dyn_dechunker_big_test);
+    tcase_add_test(rc_dechunker, test_dyn_dechunker_big_test_oversize_fail);
 
     return s;
 }

xrdp/xrdp_egfx.c

@@ -945,34 +945,8 @@ xrdp_egfx_close_response(struct xrdp_process *id, int chan_id)
 /******************************************************************************/
 /* from client */
 static int
-xrdp_egfx_data_first(struct xrdp_process *id, int chan_id,
+xrdp_egfx_data(struct xrdp_process *id, int chan_id, struct stream *s)
-                     char *data, int bytes, int total_bytes)
 {
-    struct xrdp_egfx *egfx;
-
-    LOG(LOG_LEVEL_TRACE, "xrdp_egfx_data_first: bytes %d"
-        " total_bytes %d", bytes, total_bytes);
-    egfx = id->wm->mm->egfx;
-    if (egfx->s != NULL)
-    {
-        LOG(LOG_LEVEL_ERROR, "DYNVC_DATA_FIRST PDU received while"
-            " another stream is active on channel %d", chan_id);
-        return 1;
-    }
-    make_stream(egfx->s);
-    // Caller has checked total_bytes is >= 0  and bytes is < total_bytes
-    init_stream(egfx->s, total_bytes);
-    out_uint8a(egfx->s, data, bytes);
-    return 0;
-}
-
-/******************************************************************************/
-/* from client */
-static int
-xrdp_egfx_data(struct xrdp_process *id, int chan_id, char *data, int bytes)
-{
-    int error;
-    struct stream ls;
     struct xrdp_wm *wm;
     struct xrdp_mm *mm;
     struct xrdp_egfx *egfx;
@@ -1002,32 +976,7 @@ xrdp_egfx_data(struct xrdp_process *id, int chan_id, char *data, int bytes)
         return 0;
     }
 
-    if (egfx->s == NULL)
+    return xrdp_egfx_process(egfx, s);
-    {
-        g_memset(&ls, 0, sizeof(ls));
-        ls.data = data;
-        ls.size = bytes;
-        ls.p = data;
-        ls.end = data + bytes;
-        return xrdp_egfx_process(egfx, &ls);
-    }
-    if (!s_check_rem_out(egfx->s, bytes))
-    {
-        LOG(LOG_LEVEL_ERROR, "DYNVC_DATA PDU data overflow on channel %d",
-            chan_id);
-        return 1;
-    }
-    out_uint8a(egfx->s, data, bytes);
-    if (!s_check_rem_out(egfx->s, 1))
-    {
-        s_mark_end(egfx->s);
-        egfx->s->p = egfx->s->data;
-        error = xrdp_egfx_process(egfx, egfx->s);
-        free_stream(egfx->s);
-        egfx->s = NULL;
-        return error;
-    }
-    return 0;
 }
 
 /******************************************************************************/
@@ -1048,7 +997,7 @@ xrdp_egfx_create(struct xrdp_mm *mm, struct xrdp_egfx **egfx)
     }
     procs.open_response = xrdp_egfx_open_response;
     procs.close_response = xrdp_egfx_close_response;
-    procs.data_first = xrdp_egfx_data_first;
+    procs.data_first = NULL; // Defragging handled elsewhere
     procs.data = xrdp_egfx_data;
     process = mm->wm->pro_layer;
     error = libxrdp_drdynvc_open(process->session,

xrdp/xrdp_egfx.h

@@ -127,7 +127,6 @@ struct xrdp_egfx
     int channel_id;
     int surface_id;
     int frame_id;
-    struct stream *s;
     void *user;
     struct xrdp_egfx_bulk *bulk;
     int (*caps_advertise)(void *user, int num_caps, int *version, int *flags);

xrdp/xrdp_mm.c

@@ -1044,15 +1044,6 @@ dynamic_monitor_close_response(struct xrdp_process *id, int chan_id)
     return 0;
 }
 
-/******************************************************************************/
-static int
-dynamic_monitor_data_first(struct xrdp_process *id, int chan_id,
-                           char *data, int bytes, int total_bytes)
-{
-    LOG_DEVEL(LOG_LEVEL_TRACE, "dynamic_monitor_data_first:");
-    return 0;
-}
-
 /******************************************************************************/
 int
 advance_resize_state_machine(struct xrdp_mm *mm,
@@ -1529,12 +1520,9 @@ add_resize_request_to_queue(struct xrdp_mm *self,
 
 /******************************************************************************/
 static int
-dynamic_monitor_data(struct xrdp_process *id, int chan_id,
+dynamic_monitor_data(struct xrdp_process *id, int chan_id, struct stream *s)
-                     char *data, int bytes)
 {
     int error = 0;
-    struct stream ls;
-    struct stream *s;
     int msg_type;
     int msg_length;
     struct xrdp_wm *wm;
@@ -1552,12 +1540,6 @@ dynamic_monitor_data(struct xrdp_process *id, int chan_id,
         return error;
     }
 
-    g_memset(&ls, 0, sizeof(ls));
-    ls.data = data;
-    ls.p = ls.data;
-    ls.size = bytes;
-    ls.end = ls.data + bytes;
-    s = &ls;
     in_uint32_le(s, msg_type);
     in_uint32_le(s, msg_length);
     LOG_DEVEL(LOG_LEVEL_DEBUG,
@@ -1976,7 +1958,7 @@ dynamic_monitor_initialize(struct xrdp_mm *self)
     g_memset(&d_procs, 0, sizeof(d_procs));
     d_procs.open_response = dynamic_monitor_open_response;
     d_procs.close_response = dynamic_monitor_close_response;
-    d_procs.data_first = dynamic_monitor_data_first;
+    d_procs.data_first = NULL; // Defragging handled elsewhere
     d_procs.data = dynamic_monitor_data;
     flags = 0;
     error = libxrdp_drdynvc_open(self->wm->session,
@@ -2156,65 +2138,66 @@ xrdp_mm_drdynvc_close_response(struct xrdp_process *id, int chan_id)
 /*****************************************************************************/
 /* part data from client going to channel server */
 static int
-xrdp_mm_drdynvc_data_first(struct xrdp_process *id, int chan_id, char *data,
+xrdp_mm_drdynvc_data_first(struct xrdp_process *id, int chan_id,
-                           int bytes, int total_bytes)
+                           struct stream *s, int total_bytes)
 {
     struct trans *trans;
-    struct stream *s;
+    struct stream *out_s;
     struct xrdp_wm *wm;
     int chansrv_chan_id;
+    int bytes = s_rem(s);
 
     // Size of PDU sent to chansrv
-    int pdu_size = 8 + 8 + 4 + 4 + 4 + bytes;
+    int pdu_size = 8 + 8 + 4 + 4 + bytes;
     wm = id->wm;
     trans = wm->mm->chan_trans;
-    s = trans_get_out_s(trans, pdu_size);
+    out_s = trans_get_out_s(trans, pdu_size);
-    if (s == NULL)
+    if (out_s == NULL)
     {
         return 1;
     }
-    out_uint32_le(s, 0); /* version */
+    out_uint32_le(out_s, 0); /* version */
-    out_uint32_le(s, pdu_size);
+    out_uint32_le(out_s, pdu_size);
-    out_uint32_le(s, 17); /* msg id */
+    out_uint32_le(out_s, 17); /* msg id */
-    out_uint32_le(s, pdu_size - 8);
+    out_uint32_le(out_s, pdu_size - 8);
     chansrv_chan_id = wm->mm->xr2cr_cid_map[chan_id];
-    out_uint32_le(s, chansrv_chan_id);
+    out_uint32_le(out_s, chansrv_chan_id);
-    out_uint32_le(s, bytes);
+    out_uint32_le(out_s, total_bytes);
-    out_uint32_le(s, total_bytes);
+    // Caller works out 'bytes' value from incoming stream length
-    out_uint8a(s, data, bytes);
+    out_uint8p(out_s, s->p, bytes);
-    s_mark_end(s);
+    s_mark_end(out_s);
     return trans_write_copy(trans);
 }
 
 /*****************************************************************************/
 /* data from client going to channel server */
 static int
-xrdp_mm_drdynvc_data(struct xrdp_process *id, int chan_id,
+xrdp_mm_drdynvc_data(struct xrdp_process *id, int chan_id, struct stream *s)
-                     char *data, int bytes)
 {
     struct trans *trans;
-    struct stream *s;
+    struct stream *out_s;
     struct xrdp_wm *wm;
     int chansrv_chan_id;
+    int bytes = s_rem(s);
 
     // Size of PDU sent to chansrv
-    int pdu_size = 8 + 8 + 4 + 4 + bytes;
+    int pdu_size = 8 + 8 + 4 + bytes;
     wm = id->wm;
     trans = wm->mm->chan_trans;
-    s = trans_get_out_s(trans, pdu_size);
+    out_s = trans_get_out_s(trans, pdu_size);
-    if (s == NULL)
+    if (out_s == NULL)
     {
         return 1;
     }
-    out_uint32_le(s, 0); /* version */
+    out_uint32_le(out_s, 0); /* version */
-    out_uint32_le(s, pdu_size);
+    out_uint32_le(out_s, pdu_size);
-    out_uint32_le(s, 19); /* msg id */
+    out_uint32_le(out_s, 19); /* msg id */
-    out_uint32_le(s, pdu_size - 8);
+    out_uint32_le(out_s, pdu_size - 8);
     chansrv_chan_id = wm->mm->xr2cr_cid_map[chan_id];
-    out_uint32_le(s, chansrv_chan_id);
+    out_uint32_le(out_s, chansrv_chan_id);
-    out_uint32_le(s, bytes);
+    // Caller works out 'bytes' value from incoming stream length
-    out_uint8a(s, data, bytes);
+    out_uint8p(out_s, s->p, bytes);
-    s_mark_end(s);
+    s_mark_end(out_s);
     return trans_write_copy(trans);
 }