chansrv: Reconnect ibus when the cached connection has gone stale

Fixes #3230. The static `bus` global was only ever checked for
non-NULL, not for whether the underlying connection was still alive.
If ibus disconnected (daemon restart, stale socket) or the initial
connect attempt failed, `bus` was left set to a dead/freed connection,
so every later call to xrdp_input_unicode_init() took the "already
initialized" fast path and operated on it.

- Null out bus/g_engine in the "disconnected" signal handler instead
  of leaving them dangling after g_object_unref().
- Check ibus_bus_is_connected() before trusting a cached bus, and
  tear down + reconnect if it's stale.
- Unref and clear bus on a failed connect attempt instead of leaving
  it set.
- Guard the unrefs in xrdp_input_unicode_destroy() now that bus/
  g_engine can legitimately already be NULL.

common/Makefile.am

@@ -49,6 +49,8 @@ libcommon_la_SOURCES = \
   base64.c \
   channel_defs.h \
   defines.h \
+  dechunker.c \
+  dechunker.h \
   fifo.c \
   fifo.h \
   file.c \

common/dechunker.c

@@ -0,0 +1,549 @@
+/**
+ * xrdp: A Remote Desktop Protocol server.
+ *
+ * Copyright (C) Jay Sorg 2006-2026
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+/**
+ * @file dechunker.c
+ * @brief Dechunker functions for chunks on virtual channels - definitions
+ * @author Matt Burt
+ *
+ */
+
+#if defined(HAVE_CONFIG_H)
+#include <config_ac.h>
+#endif
+
+#include "dechunker.h"
+#include "parse.h"
+#include "os_calls.h"
+#include "string_calls.h"
+
+enum vc_dechunker_state
+{
+    E_NO_DATA = 0,
+    E_READING,
+    E_DATA,
+    E_SKIPPING
+};
+
+enum vc_chunk_type
+{
+    CT_INTERMEDIATE = 0,
+    CT_FIRST = 1, // CHANNEL_FLAG_FIRST
+    CT_LAST = 2,  // CHANNEL_FLAG_LAST
+    CT_FIRST_LAST = 3 // CHANNEL_FLAG_FIRST | CHANNEL_FLAG_LAST
+};
+
+struct vc_dechunker
+{
+    char name[64];
+    int max_chunk_size;
+    enum vc_dechunker_state state;
+    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_VC_CHUNK_SIZE_LOWER_LIMIT = 50
+};
+/*****************************************************************************/
+struct vc_dechunker *
+vc_dechunker_init(const char *chan_name, int max_chunk_size)
+{
+    struct vc_dechunker *self = NULL;
+    if (chan_name == NULL)
+    {
+        LOG(LOG_LEVEL_ERROR, "vc_dechunker_init() called with no channel name");
+    }
+    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);
+    }
+    else if ((self = g_new(struct vc_dechunker, 1)) == NULL)
+    {
+        LOG(LOG_LEVEL_ERROR, "Dechunker: no memory for %s", chan_name);
+    }
+    else
+    {
+        strlcpy(self->name, chan_name, sizeof(self->name));
+        self->max_chunk_size = max_chunk_size;
+        self->state = E_NO_DATA;
+        self->reassembly_s = NULL;
+    }
+
+    return self;
+}
+
+/*****************************************************************************/
+void
+vc_dechunker_free(struct vc_dechunker *self)
+{
+    if (self != NULL)
+    {
+        free_stream(self->reassembly_s);
+        free(self);
+    }
+}
+
+/*****************************************************************************/
+static void
+vc_dechunker_reset(struct vc_dechunker *self)
+{
+    if (self != NULL)
+    {
+        free_stream(self->reassembly_s);
+        self->reassembly_s = NULL;
+        self->state = E_NO_DATA;
+    }
+}
+
+/*****************************************************************************/
+static void
+log_unexpected_vc_chunk_type(struct vc_dechunker *self,
+                             enum vc_chunk_type ct)
+{
+    static const char *chunk_type_str[4] =
+    {
+        "CT_INTERMEDIATE",
+        "CT_FIRST",
+        "CT_LAST",
+        "CT_FIRST_LAST"
+    };
+
+    int index = (int)ct & 3; // Guarantee to be 0..3
+    LOG (LOG_LEVEL_ERROR,
+         "Dechunker: unexpected chunk type %s received on %s",
+         chunk_type_str[index], self->name);
+    self->state = E_SKIPPING; // Look for the next PDU
+}
+
+/*****************************************************************************/
+static enum vc_dechunker_status
+handle_no_data_state(struct vc_dechunker *self,
+                     struct stream *s,
+                     int chunk_size,
+                     int total_size,
+                     enum vc_chunk_type ct)
+{
+    enum vc_dechunker_status rv = E_VC_ERROR;
+    switch (ct)
+    {
+        case CT_FIRST:
+            // See [MS-RDPBCGR] 3.1.5.2.1 Sending of Virtual Channel PDU
+            if (total_size <= self->max_chunk_size)
+            {
+                LOG (LOG_LEVEL_ERROR,
+                     "Dechunker: short first chunk received on %s",
+                     self->name);
+                self->state = E_SKIPPING;
+            }
+            else if (chunk_size >= total_size)
+            {
+                LOG (LOG_LEVEL_ERROR,
+                     "Dechunker: malformed first chunk received on %s",
+                     self->name);
+                self->state = E_SKIPPING;
+            }
+            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,
+                         "Dechunker: out-of-memory on %s",
+                         self->name);
+                    self->state = E_SKIPPING;
+                }
+                else
+                {
+                    out_uint8p(self->reassembly_s, s->p, chunk_size);
+                    in_uint8s(s, chunk_size);
+                    self->state = E_READING;
+                    rv = E_VC_IN_PROGRESS;
+                }
+            }
+            break;
+
+        case CT_FIRST_LAST:
+            rv = E_VC_INLINE_CHUNK;
+            break;
+
+        default:
+            log_unexpected_vc_chunk_type(self, ct);
+            self->state = E_SKIPPING;
+    }
+
+    return rv;
+}
+
+/*****************************************************************************/
+static enum vc_dechunker_status
+handle_reading_state(struct vc_dechunker *self,
+                     struct stream *s,
+                     int chunk_size,
+                     int total_size,
+                     enum vc_chunk_type ct)
+{
+    enum vc_dechunker_status rv = E_VC_ERROR;
+    if (ct == CT_INTERMEDIATE || ct == CT_LAST)
+    {
+        /* Data to add to the reassembly stream
+         *
+         * [MS-RDPBCGR] 3.1.5.2.2.1 imposes no requirement to check
+         * the total_size field is consistent between chunks. Only
+         * the total_size value for CT_FIRST is important
+         */
+        if (chunk_size > s_rem_out(self->reassembly_s))
+        {
+            LOG (LOG_LEVEL_ERROR,
+                 "Dechunker: oversized chunk received on %s",
+                 self->name);
+            self->state = E_SKIPPING;
+        }
+        else
+        {
+            out_uint8p(self->reassembly_s, s->p, chunk_size);
+            in_uint8s(s, chunk_size);
+            if (ct == CT_LAST)
+            {
+                if (s_rem_out(self->reassembly_s) == 0)
+                {
+                    // Make the stream ready for reading
+                    s_mark_end(self->reassembly_s);
+                    self->reassembly_s->p = self->reassembly_s->data;
+                    // 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
+                {
+                    LOG (LOG_LEVEL_ERROR,
+                         "Dechunker: undersized last chunk received on %s",
+                         self->name);
+                    self->state = E_SKIPPING;
+                }
+            }
+            else
+            {
+                rv = E_VC_IN_PROGRESS;
+            }
+        }
+    }
+    else
+    {
+        log_unexpected_vc_chunk_type(self, ct);
+    }
+
+    return rv;
+}
+
+/*****************************************************************************/
+enum vc_dechunker_status
+vc_dechunker_process_chunk(struct vc_dechunker *self,
+                           struct stream *s, int flags,
+                           int total_size)
+{
+    enum vc_dechunker_status rv = E_VC_ERROR;
+    enum vc_chunk_type ct = (enum vc_chunk_type)(flags & 3);
+    int chunk_size = s ? s_rem(s) : 0; // Chunk is remainder of stream
+
+    if (self == NULL || s == NULL)
+    {
+        ; // Nothing to do
+    }
+    else if (total_size < 0)
+    {
+        LOG (LOG_LEVEL_ERROR,
+             "Dechunker: out-of-range length received on %s",
+             self->name);
+        self->state = E_SKIPPING;
+    }
+    else if (chunk_size > self->max_chunk_size)
+    {
+        // [MS-RDPBCGR] 2.2.6.1
+        // > ... This field MUST NOT be larger than CHANNEL_CHUNK_size
+        // > (1600) bytes in size unless the maximum virtual channel
+        // > chunk size is specified in the optional VCChunkSize field
+        // > of the Virtual Channel Capability Set (section 2.2.7.1.10).
+        LOG (LOG_LEVEL_ERROR,
+             "Dechunker: oversize chunk received on %s (%d octets)",
+             self->name, chunk_size);
+        self->state = E_SKIPPING;
+    }
+    else
+    {
+        // Check for a restart after an error
+        if (self->state == E_SKIPPING)
+        {
+            switch (ct)
+            {
+                case CT_FIRST:
+                case CT_FIRST_LAST:
+                    // Clean up the dechunker and start again
+                    vc_dechunker_reset(self);
+                    break;
+                default:
+                    break;
+            }
+        }
+
+        switch (self->state)
+        {
+            case E_NO_DATA:
+                rv = handle_no_data_state(self, s, chunk_size,
+                                          total_size, ct);
+                break;
+
+            case E_READING:
+                rv = handle_reading_state(self, s, chunk_size,
+                                          total_size, ct);
+                break;
+
+            case E_DATA:
+                // If we get here, we've not cleared the existing buffer.
+                // This is a serious problem and we continue returning
+                // a error until the buffer is cleared.
+                LOG(LOG_LEVEL_ALWAYS,
+                    "Dechunker: unprocessed PDU on %s", self->name);
+                break;
+
+            case E_SKIPPING:
+                rv = E_VC_IN_PROGRESS; // Ignore this chunk
+                break;
+
+            default:
+                // Shouldn't get here.
+                LOG (LOG_LEVEL_ERROR,
+                     "Dechunker: called when %s has an unknown state %d",
+                     self->name, (int)self->state);
+                self->state = E_SKIPPING;
+        }
+    }
+
+    return rv;
+}
+
+/*****************************************************************************/
+struct stream *
+vc_dechunker_get_stream(struct vc_dechunker *self)
+{
+    struct stream *s;
+    const char *name = (self != NULL) ? self->name : "<unknown>";
+    if (self == NULL || self->state != E_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!
+        vc_dechunker_reset(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

@@ -0,0 +1,189 @@
+/**
+ * xrdp: A Remote Desktop Protocol server.
+ *
+ * Copyright (C) Jay Sorg 2004-2026
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+/**
+ * @file dechunker.h
+ * @brief Dechunker functions for chunks on virtual channels - declarations
+ *
+ * Many places in xrdp need to dechunk data received from virtual channels.
+ * This process is described in [MS-RDPBCGR] 3.1.5.2.1 (sending) and
+ * 3.1.5.2.2.1 (reassembly).
+ *
+ *
+ * @author Matt Burt
+ */
+
+#if !defined(DECHUNKER_H)
+#define DECHUNKER_H
+
+struct stream;
+
+/* Private types */
+struct vc_dechunker;   // static virtual channel dechunker
+struct dyn_dechunker;  // dynamic channel dechunker
+
+/**
+ * Returned from vc_dechunker_process_chunk()
+ */
+enum vc_dechunker_status
+{
+    E_VC_INLINE_CHUNK = 0, ///< This chunk is complete in itself
+    E_VC_IN_PROGRESS,      ///< The dechunker is processing chunks
+    E_VC_READY,            ///< A dechunked stream is now complete
+    E_VC_ERROR             ///< An error occurred (logged)
+};
+
+/**
+ * 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
+ * @return vc_dechunker
+ */
+struct vc_dechunker *
+vc_dechunker_init(const char *chan_name, int max_chunk_size);
+
+/**
+ * Free a static virtual channel dechunker
+ * @param self vc dechunker to free
+ */
+void
+vc_dechunker_free(struct vc_dechunker *self);
+
+/**
+ * Process a static virtual channel chunk
+ *
+ * @param self dechunker
+ * @param s Stream for chunk, positioned at start of chunk
+ * @param flags from CHANNEL_PDU_HEADER
+ * @param total_size length from CHANNEL_PDU_HEADER
+ * @return status of dechunker
+ *
+ * If E_VC_ERROR is returned, the dechunker will ignore further PDUs
+ * until the start of the next one is detected. This can be used to
+ * recover from chunking errors without losing the channel entirely. It
+ * is up to the caller whether to treat a dechunking error as fatal for
+ * the channel or not.
+ */
+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 static virtual dechunker
+ *
+ * @param self static virtual 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 *
+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,14 +143,17 @@ 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
+
 /* channel */
 struct xrdp_channel
 {
@@ -158,8 +161,8 @@ struct xrdp_channel
     struct xrdp_mcs *mcs_layer;
     int drdynvc_channel_id;
     int drdynvc_state;
-    struct stream *s;
+    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

@@ -23,7 +23,7 @@
 #endif
 
 #include "libxrdp.h"
-#include "parse.h"
+#include "dechunker.h"
 #include "string_calls.h"
 #include "xrdp_channel.h"
 
@@ -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;
     }
-    free_stream(self->s);
+    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;
         }
-        in_uint32_le(s, total_bytes); /* Length */
-    }
-    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
+        LOG_DEVEL(LOG_LEVEL_TRACE, "Received [MS-RDPEDYC] DYNVC_DATA_FIRST "
-    if (total_bytes < 1590 || bytes > total_bytes)
+                  "ChannelId %d, Length %d, Data (omitted from the log)",
-    {
+                  chan_id, total_bytes);
-        LOG(LOG_LEVEL_ERROR,
+
-            "Badly formed DYNVC_DATA_FIRST PDU received on dynamic channel %d",
+        if (chan_id >= DRDYNVC_CHANNEL_COUNT)
-            chan_id);
+        {
-        return 1;
+            LOG(LOG_LEVEL_ERROR, "Received [MS-RDPEDYC] DYNVC_DATA_FIRST "
+                "for an invalid channel id %d", chan_id);
+            rv = 1;
+        }
+        else
+        {
+            struct xrdp_drdynvc *drdynvc = self->drdynvcs + chan_id;
+
+            if (drdynvc->data_first != NULL)
+            {
+                // Caller has requested to defragment PDUs themself
+                struct xrdp_session *session =
+                        self->sec_layer->rdp_layer->session;
+                rv = drdynvc->data_first(session->id, chan_id, s, total_bytes);
+            }
+            else
+            {
+                // 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)
+                        {
+                            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;
+                }
+            }
+        }
     }
 
-    session = self->sec_layer->rdp_layer->session;
+    return rv;
-    if (chan_id > 255)
+
-    {
+short_pdu:
-        LOG(LOG_LEVEL_ERROR, "Received [MS-RDPEDYC] DYNVC_DATA_FIRST for an "
+    LOG(LOG_LEVEL_ERROR, "[MS-RDPEDYC] DYNVC_DATA_FIRST is too short");
-            "invalid channel id. Max allowed 255, received %d", chan_id);
+    return 1;
-        return 1;
-    }
-    drdynvc = self->drdynvcs + chan_id;
-    if (drdynvc->data_first != NULL)
-    {
-        return drdynvc->data_first(session->id, chan_id, s->p,
-                                   bytes, total_bytes);
-    }
-    LOG_DEVEL(LOG_LEVEL_WARNING, "Dynamic Virtual Channel %s (%d): "
-              "callback 'data_first' is NULL",
-              XRDP_DRDYNVC_CHANNEL_ID_TO_NAME(self, chan_id),
-              chan_id);
-    return 0;
 }
 
 /*****************************************************************************/
@@ -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)",
-              chan_id, bytes);
-    session = self->sec_layer->rdp_layer->session;
-    if (chan_id > 255)
     {
-        LOG(LOG_LEVEL_ERROR, "Received DYNVC_DATA PDU for an invalid "
+        LOG_DEVEL(LOG_LEVEL_TRACE, "Received [MS-RDPEDYC] DYNVC_DATA "
-            "channel id. channel id %d", chan_id);
+                  "ChannelId %d, Length %d, Data (omitted from the log)",
-        return 1;
+                  chan_id, s_rem(s));
+        if (chan_id >= DRDYNVC_CHANNEL_COUNT)
+        {
+            LOG(LOG_LEVEL_ERROR, "Received DYNVC_DATA PDU for an invalid "
+                "channel id %d", chan_id);
+            rv = 1;
+        }
+        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)
+                {
+                    struct xrdp_session *session =
+                            self->sec_layer->rdp_layer->session;
+                    rv = drdynvc->data(session->id, chan_id, ls);
+                }
+
+                if (free_ls)
+                {
+                    free_stream(ls);
+                }
+            }
+        }
     }
-    drdynvc = self->drdynvcs + chan_id;
+    return rv;
-    if (drdynvc->data != NULL)
-    {
-        return drdynvc->data(session->id, chan_id, s->p, bytes);
-    }
-    LOG_DEVEL(LOG_LEVEL_WARNING, "Dynamic Virtual Channel %s (%d): "
-              "callback 'data' is NULL",
-              XRDP_DRDYNVC_CHANNEL_ID_TO_NAME(self, chan_id),
-              chan_id);
-    return 0;
 }
 
 /*****************************************************************************/
@@ -550,11 +656,12 @@ xrdp_channel_process_drdynvc(struct xrdp_channel *self,
                              struct stream *s)
 {
     int total_length;
-    int length;
     int flags;
     int cmd;
     int rv;
-    struct stream *ls;
+    struct stream *ls = NULL;
+    int free_ls = 0;
+    enum vc_dechunker_status dechunker_status;
 
     if (!s_check_rem_and_log(s, 8, "Parsing [MS-RDPBCGR] CHANNEL_PDU_HEADER"))
     {
@@ -564,65 +671,32 @@ xrdp_channel_process_drdynvc(struct xrdp_channel *self,
     in_uint32_le(s, flags);        /* flags */
     LOG_DEVEL(LOG_LEVEL_TRACE, "Received header [MS-RDPBCGR] CHANNEL_PDU_HEADER "
               "length %d, flags 0x%8.8x", total_length, flags);
-    ls = NULL;
+    dechunker_status = vc_dechunker_process_chunk(
-    switch (flags & 3)
+                           self->drdynvc_dc,
+                           s, flags, total_length);
+    switch (dechunker_status)
     {
-        case 0: /* not first chunk and not last chunk */
+        case E_VC_INLINE_CHUNK:
-            length = (int) (s->end - s->p);
-            LOG_DEVEL(LOG_LEVEL_TRACE, "Received [MS-RDPBCGR] data chunk (middle) "
-                      "length %d", length);
-            if (length > s_rem_out(self->s))
-            {
-                LOG(LOG_LEVEL_ERROR, "[MS-RDPBCGR] Data chunk length is bigger than "
-                    "the remaining chunk buffer size. length %d, remaining %d",
-                    length, s_rem_out(self->s));
-                return 1;
-            }
-            out_uint8a(self->s, s->p, length); /* append data to chunk buffer */
-            in_uint8s(s, length);              /* virtualChannelData */
-            return 0;
-        case 1: /* CHANNEL_FLAG_FIRST */
-            free_stream(self->s);
-            make_stream(self->s);
-            init_stream(self->s, total_length);
-            length = (int) (s->end - s->p);
-            LOG_DEVEL(LOG_LEVEL_TRACE, "Received [MS-RDPBCGR] data chunk (first) "
-                      "length %d", length);
-            if (length > s_rem_out(self->s))
-            {
-                LOG(LOG_LEVEL_ERROR, "[MS-RDPBCGR] Data chunk length is bigger than "
-                    "the remaining chunk buffer size. length %d, remaining %d",
-                    length, s_rem_out(self->s));
-                return 1;
-            }
-            out_uint8a(self->s, s->p, length); /* append data to chunk buffer */
-            in_uint8s(s, length);              /* virtualChannelData */
-            return 0;
-        case 2: /* CHANNEL_FLAG_LAST */
-            length = (int) (s->end - s->p);
-            LOG_DEVEL(LOG_LEVEL_TRACE, "Received [MS-RDPBCGR] data chunk (last) "
-                      "length %d", length);
-            if (length > s_rem_out(self->s))
-            {
-                LOG(LOG_LEVEL_ERROR, "[MS-RDPBCGR] Data chunk length is bigger than "
-                    "the remaining chunk buffer size. length %d, remaining %d",
-                    length, s_rem_out(self->s));
-                return 1;
-            }
-            out_uint8a(self->s, s->p, length); /* append data to chunk buffer */
-            in_uint8s(s, length);              /* virtualChannelData */
-            s_mark_end(self->s);
-            self->s->p = self->s->data;
-            ls = self->s;
-            break;
-        case 3: /* CHANNEL_FLAG_FIRST and CHANNEL_FLAG_LAST */
-            LOG_DEVEL(LOG_LEVEL_TRACE, "Received [MS-RDPBCGR] data chunk (first and last) "
-                      "length %d", total_length);
             ls = s;
             break;
+
+        case E_VC_IN_PROGRESS:
+            return 0;
+            break;
+
+        case E_VC_READY:
+            ls = vc_dechunker_get_stream(self->drdynvc_dc);
+            // We now own the stream, so must delete it
+            free_ls = 1;
+            break;
+
+        case E_VC_ERROR:
+            // Error has been logged
+            return 1;
+
         default:
-            LOG(LOG_LEVEL_ERROR, "Received [MS-RDPBCGR] data chunk with "
+            LOG(LOG_LEVEL_ERROR, "Dechunker returned unknown error %d",
-                "unknown flag 0x%8.8x", (int) (flags & 3));
+                (int)dechunker_status);
             return 1;
     }
     if (ls == NULL)
@@ -657,6 +731,10 @@ xrdp_channel_process_drdynvc(struct xrdp_channel *self,
                 "unknown command 0x%2.2x", cmd);
             break;
     }
+    if (free_ls)
+    {
+        free_stream(ls);
+    }
     return rv;
 }
 
@@ -806,6 +884,13 @@ xrdp_channel_drdynvc_start(struct xrdp_channel *self)
                 DRDYNVC_SVC_CHANNEL_NAME);
             rv = -1;
         }
+        else if ((self->drdynvc_dc =
+                      vc_dechunker_init(DRDYNVC_SVC_CHANNEL_NAME,
+                                        CHANNEL_CHUNK_LENGTH)) == NULL)
+        {
+            LOG(LOG_LEVEL_ERROR, "No memory");
+            rv = -1;
+        }
         else
         {
             self->drdynvc_channel_id = (dci->chanid - MCS_GLOBAL_CHANNEL) - 1;
@@ -841,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);
@@ -848,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 */
@@ -857,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 */
@@ -886,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;
 }
 
 /*****************************************************************************/
@@ -914,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;
     }
@@ -984,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)
@@ -1056,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
+    {
+        // 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)
         {
-            return 1;
+            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 != NULL)
+    if (drdynvc->data_first != NULL)
     {
-        if (drdynvc->data(chan_id, data, bytes) != 0)
+        // 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)
         {
-            return 1;
+            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;
         }
     }
-    return 0;
+
+    if (ls != NULL)
+    {
+        if (drdynvc->data != NULL)
+        {
+            rv = drdynvc->data(chan_id, ls);
+        }
+        if (free_ls)
+        {
+            free_stream(ls);
+        }
+    }
+    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].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

sesman/chansrv/input_ibus.c

@@ -30,7 +30,7 @@
 static IBusBus *bus;
 static IBusEngine *g_engine;
 /* This is the engine name enabled before unicode engine enabled */
-static const gchar *last_input_name;
+static gchar *last_input_name;
 static int id = 0;
 
 static int
@@ -39,27 +39,38 @@ xrdp_input_enable(void)
     IBusEngineDesc *desc;
     const gchar *name;
 
-    if (last_input_name)
-    {
-        /* already enabled */
-        return 0;
-    }
-
     if (!bus)
     {
         LOG(LOG_LEVEL_ERROR, "xrdp_ibus_init: input method switched failed, ibus not connected");
         return 1;
     }
 
+    /* Re-check the current global engine on every call rather than
+     * trusting a one-time flag: ibus (particularly with
+     * use_global_engine disabled, which is common) can silently swap
+     * the active engine back to the user's own IME between calls, and
+     * we need to notice that and reassert XrdpIme rather than keep
+     * committing text to an engine that's no longer active. */
     desc = ibus_bus_get_global_engine(bus);
-    name = ibus_engine_desc_get_name (desc);
+    name = desc ? ibus_engine_desc_get_name(desc) : NULL;
-    if (!g_ascii_strcasecmp(name, "XrdpIme"))
+    if (name && !g_ascii_strcasecmp(name, "XrdpIme"))
     {
+        g_object_unref(desc);
         return 0;
     }
 
-    /* remember user's input method, will switch back when disconnect */
+    if (!last_input_name && name)
-    last_input_name = name;
+    {
+        /* remember user's original input method (first time only), will
+         * switch back when disconnected. Copy the name out since it's
+         * owned by desc, which we're about to unref. */
+        last_input_name = g_strdup(name);
+    }
+
+    if (desc)
+    {
+        g_object_unref(desc);
+    }
 
     if (!ibus_bus_set_global_engine(bus, "XrdpIme"))
     {
@@ -105,8 +116,18 @@ static void
 xrdp_input_ibus_disconnect(IBusEngine *engine)
 {
     LOG(LOG_LEVEL_INFO, "xrdp_ibus_engine_disable: IM disabled");
-    g_object_unref(g_engine);
+    if (g_engine)
-    g_object_unref(bus);
+    {
+        g_object_unref(g_engine);
+        g_engine = NULL;
+    }
+    if (bus)
+    {
+        g_object_unref(bus);
+        bus = NULL;
+    }
+    g_free(last_input_name);
+    last_input_name = NULL;
 }
 
 static gboolean
@@ -125,12 +146,13 @@ xrdp_input_ibus_create_engine(IBusFactory *factory,
                               gpointer user_data)
 {
     IBusEngine *engine;
-    gchar *path = g_strdup_printf("/org/freedesktop/IBus/Engine/%i", 1);
+    gchar *path = g_strdup_printf("/org/freedesktop/IBus/Engine/%i", ++id);
     engine = ibus_engine_new(engine_name,
                              path,
                              ibus_bus_get_connection(bus));
 
-    LOG(LOG_LEVEL_DEBUG, "xrdp_input_ibus_create_engine: Creating IM Engine with name:%s and id:%d\n", engine_name, ++id);
+    LOG(LOG_LEVEL_DEBUG, "xrdp_input_ibus_create_engine: Creating IM Engine with name:%s and id:%d\n", engine_name, id);
+    g_free(path);
 
     g_signal_connect(engine, "process-key-event", G_CALLBACK(engine_process_key_event_cb), NULL);
     g_signal_connect(engine, "enable", G_CALLBACK(xrdp_input_ibus_engine_enable), NULL);
@@ -192,15 +214,22 @@ int
 xrdp_input_unicode_destroy(void)
 {
     LOG(LOG_LEVEL_DEBUG, "xrdp_input_unicode_destory: ibus input is under destory");
-    if (last_input_name)
+    if (last_input_name && bus)
     {
         LOG(LOG_LEVEL_INFO, "xrdp_input_unicode_destory: ibus engine rolling back to origin: %s", last_input_name);
         ibus_bus_set_global_engine(bus, last_input_name);
     }
 
-    g_object_unref(g_engine);
+    if (g_engine)
-    g_object_unref(bus);
+    {
+        g_object_unref(g_engine);
+    }
+    if (bus)
+    {
+        g_object_unref(bus);
+    }
 
+    g_free(last_input_name);
     last_input_name = NULL;
     bus = NULL;
     g_engine = NULL;
@@ -213,9 +242,27 @@ xrdp_input_unicode_init(void)
 {
     if (bus)
     {
-        /* Already initialized, just re-enable it */
+        if (ibus_bus_is_connected(bus))
-        xrdp_input_enable();
+        {
-        return 0;
+            /* Already initialized, just re-enable it */
+            xrdp_input_enable();
+            return 0;
+        }
+
+        /* The bus is stale (e.g. the ibus daemon restarted and left a
+         * dead connection behind). Tear it down so we reconnect below
+         * instead of operating on a dead connection. */
+        LOG(LOG_LEVEL_WARNING,
+            "xrdp_ibus_init: existing iBus connection is stale, reconnecting");
+        if (g_engine)
+        {
+            g_object_unref(g_engine);
+            g_engine = NULL;
+        }
+        g_object_unref(bus);
+        bus = NULL;
+        g_free(last_input_name);
+        last_input_name = NULL;
     }
 
     /* Wait because the ibus daemon may not be ready on first login */
@@ -242,6 +289,8 @@ xrdp_input_unicode_init(void)
     if (!ibus_bus_is_connected(bus))
     {
         LOG(LOG_LEVEL_ERROR, "xrdp_ibus_init: Connect to iBus failed");
+        g_object_unref(bus);
+        bus = NULL;
         return 1;
     }
 
@@ -251,6 +300,9 @@ xrdp_input_unicode_init(void)
 
     if (!ibus_bus_get_global_engine(bus))
     {
+        /* The bus connection itself is fine (and is now owned by the
+         * ibus main loop thread we just started), so leave it in place
+         * rather than tearing it down here. */
         LOG(LOG_LEVEL_ERROR, "xrdp_ibus_init: failed to get origin global engine");
         return 1;
     }

tests/common/Makefile.am

@@ -14,6 +14,7 @@ check_PROGRAMS = test_common
 test_common_SOURCES = \
     test_common.h \
     test_common_main.c \
+    test_dechunker.c \
     test_fifo_calls.c \
     test_list_calls.c \
     test_list16_calls.c \

tests/common/test_common.h

@@ -7,6 +7,7 @@
 char *
 bin_to_hex(const char *input, int length);
 
+Suite *make_suite_test_dechunker(void);
 Suite *make_suite_test_fifo(void);
 Suite *make_suite_test_list(void);
 Suite *make_suite_test_list16(void);

tests/common/test_common_main.c

@@ -41,24 +41,75 @@ bin_to_hex(const char *input, int length)
     return result;
 }
 
+static int
+run_suite(const char *test_name)
+{
+    const char *env = getenv("TEST_NAME");
+    return (env == NULL || strcmp(env, test_name) == 0);
+}
+
 int main (void)
 {
     int number_failed;
     SRunner *sr;
 
-    sr = srunner_create (make_suite_test_fifo());
+    sr = srunner_create (NULL);
-    srunner_add_suite(sr, make_suite_test_list());
+    if (run_suite("dechunker"))
-    srunner_add_suite(sr, make_suite_test_list16());
+    {
-    srunner_add_suite(sr, make_suite_test_parse());
+        srunner_add_suite(sr, make_suite_test_dechunker());
-    srunner_add_suite(sr, make_suite_test_set_int());
+    }
-    srunner_add_suite(sr, make_suite_test_string());
+    if (run_suite("fifo"))
-    srunner_add_suite(sr, make_suite_test_string_unicode());
+    {
-    srunner_add_suite(sr, make_suite_test_os_calls());
+        srunner_add_suite(sr, make_suite_test_fifo());
-    srunner_add_suite(sr, make_suite_test_ssl_calls());
+    }
-    srunner_add_suite(sr, make_suite_test_base64());
+    if (run_suite("list"))
-    srunner_add_suite(sr, make_suite_test_guid());
+    {
-    srunner_add_suite(sr, make_suite_test_scancode());
+        srunner_add_suite(sr, make_suite_test_list());
-    srunner_add_suite(sr, make_suite_test_timers());
+    }
+    if (run_suite("list16"))
+    {
+        srunner_add_suite(sr, make_suite_test_list16());
+    }
+    if (run_suite("parse"))
+    {
+        srunner_add_suite(sr, make_suite_test_parse());
+    }
+    if (run_suite("set_int"))
+    {
+        srunner_add_suite(sr, make_suite_test_set_int());
+    }
+    if (run_suite("string"))
+    {
+        srunner_add_suite(sr, make_suite_test_string());
+    }
+    if (run_suite("unicode"))
+    {
+        srunner_add_suite(sr, make_suite_test_string_unicode());
+    }
+    if (run_suite("os_calls"))
+    {
+        srunner_add_suite(sr, make_suite_test_os_calls());
+    }
+    if (run_suite("ssl_calls"))
+    {
+        srunner_add_suite(sr, make_suite_test_ssl_calls());
+    }
+    if (run_suite("base64"))
+    {
+        srunner_add_suite(sr, make_suite_test_base64());
+    }
+    if (run_suite("guid"))
+    {
+        srunner_add_suite(sr, make_suite_test_guid());
+    }
+    if (run_suite("scancode"))
+    {
+        srunner_add_suite(sr, make_suite_test_scancode());
+    }
+    if (run_suite("timers"))
+    {
+        srunner_add_suite(sr, make_suite_test_timers());
+    }
 
     srunner_set_tap(sr, "-");
     /*

tests/common/test_dechunker.c

@@ -0,0 +1,983 @@
+/**
+ * xrdp: A Remote Desktop Protocol server.
+ *
+ * Copyright (C) Jay Sorg 2006-2026
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+/**
+ * @file test_dechunker.c
+ * @brief Test functiond for dechunker code
+ * @author Matt Burt
+ *
+ */
+
+
+#if defined(HAVE_CONFIG_H)
+#include "config_ac.h"
+#endif
+
+#include "dechunker.h"
+#include "ms-rdpbcgr.h"
+#include "os_calls.h"
+#include "parse.h"
+
+#include "test_common.h"
+
+// Chapter 1 of Mary Shelley's Frankenstein (thanks to Project Gutenberg)
+static const char frankenstein[] =
+    "Letter 1\n\n"
+    "To Mrs. Saville, England.\n\n"
+    "St. Petersburgh, Dec. 11th, 17—.\n"
+    "You will rejoice to hear that no disaster has accompanied the "
+    "commencement of an enterprise which you have regarded with such evil "
+    "forebodings. I arrived here yesterday, and my first task is to assure "
+    "my dear sister of my welfare and increasing confidence in the success "
+    "of my undertaking.\n"
+    "I am already far north of London, and as I walk in the streets of "
+    "Petersburgh, I feel a cold northern breeze play upon my cheeks, which "
+    "braces my nerves and fills me with delight. Do you understand this "
+    "feeling? This breeze, which has travelled from the regions towards which "
+    "I am advancing, gives me a foretaste of those icy climes. Inspirited "
+    "by this wind of promise, my daydreams become more fervent and vivid. I "
+    "try in vain to be persuaded that the pole is the seat of frost and "
+    "desolation; it ever presents itself to my imagination as the region "
+    "of beauty and delight. There, Margaret, the sun is for ever visible, "
+    "its broad disk just skirting the horizon and diffusing a perpetual "
+    "splendour. There—for with your leave, my sister, I will put some trust "
+    "in preceding navigators—there snow and frost are banished; and, sailing "
+    "over a calm sea, we may be wafted to a land surpassing in wonders and "
+    "in beauty every region hitherto discovered on the habitable globe. Its "
+    "productions and features may be without example, as the phenomena of the "
+    "heavenly bodies undoubtedly are in those undiscovered solitudes. What "
+    "may not be expected in a country of eternal light? I may there discover "
+    "the wondrous power which attracts the needle and may regulate a thousand "
+    "celestial observations that require only this voyage to render their "
+    "seeming eccentricities consistent for ever. I shall satiate my ardent "
+    "curiosity with the sight of a part of the world never before visited, "
+    "and may tread a land never before imprinted by the foot of man. These "
+    "are my enticements, and they are sufficient to conquer all fear of danger "
+    "or death and to induce me to commence this laborious voyage with the joy "
+    "a child feels when he embarks in a little boat, with his holiday mates, "
+    "on an expedition of discovery up his native river. But supposing all "
+    "these conjectures to be false, you cannot contest the inestimable benefit "
+    "which I shall confer on all mankind, to the last generation, by "
+    "discovering a passage near the pole to those countries, to reach which at "
+    "present so many months are requisite; or by ascertaining the secret of "
+    "the magnet, which, if at all possible, can only be effected by an "
+    " undertaking such as mine.\n"
+    "These reflections have dispelled the agitation with which I began my "
+    "letter, and I feel my heart glow with an enthusiasm which elevates me "
+    "to heaven, for nothing contributes so much to tranquillise the mind as "
+    "a steady purpose—a point on which the soul may fix its intellectual "
+    "eye. This expedition has been the favourite dream of my early years. I "
+    "have read with ardour the accounts of the various voyages which have "
+    "been made in the prospect of arriving at the North Pacific Ocean through "
+    "the seas which surround the pole. You may remember that a history of "
+    "all the voyages made for purposes of discovery composed the whole of "
+    "our good Uncle Thomas’ library. My education was neglected, yet I was "
+    "passionately fond of reading. These volumes were my study day and night, "
+    "and my familiarity with them increased that regret which I had felt, as "
+    "a child, on learning that my father’s dying injunction had forbidden "
+    "my uncle to allow me to embark in a seafaring life.\n"
+    "These visions faded when I perused, for the first time, those poets whose "
+    "effusions entranced my soul and lifted it to heaven. I also became a poet "
+    "and for one year lived in a paradise of my own creation; I imagined that "
+    "I also might obtain a niche in the temple where the names of Homer and "
+    "Shakespeare are consecrated. You are well acquainted with my failure and "
+    "how heavily I bore the disappointment. But just at that time I inherited "
+    "the fortune of my cousin, and my thoughts were turned into the channel "
+    "of their earlier bent.\n"
+    "Six years have passed since I resolved on my present undertaking. I can, "
+    "even now, remember the hour from which I dedicated myself to this great "
+    "enterprise. I commenced by inuring my body to hardship. I accompanied "
+    "the whale-fishers on several expeditions to the North Sea; I voluntarily "
+    "endured cold, famine, thirst, and want of sleep; I often worked harder "
+    "than the common sailors during the day and devoted my nights to the "
+    "study of mathematics, the theory of medicine, and those branches of "
+    "physical science from which a naval adventurer might derive the greatest "
+    "practical advantage. Twice I actually hired myself as an under-mate in "
+    "a Greenland whaler, and acquitted myself to admiration. I must own I "
+    "felt a little proud when my captain offered me the second dignity in "
+    "the vessel and entreated me to remain with the greatest earnestness, "
+    "so valuable did he consider my services.\n"
+    "And now, dear Margaret, do I not deserve to accomplish some great "
+    "purpose? My life might have been passed in ease and luxury, but I "
+    "preferred glory to every enticement that wealth placed in my path. Oh, "
+    "that some encouraging voice would answer in the affirmative! My courage "
+    "and my resolution is firm; but my hopes fluctuate, and my spirits are "
+    "often depressed. I am about to proceed on a long and difficult voyage, "
+    "the emergencies of which will demand all my fortitude: I am required not "
+    "only to raise the spirits of others, but sometimes to sustain my own, "
+    "when theirs are failing.\n"
+    "This is the most favourable period for travelling in Russia. They fly "
+    "quickly over the snow in their sledges; the motion is pleasant, and, in "
+    "my opinion, far more agreeable than that of an English stagecoach. The "
+    "cold is not excessive, if you are wrapped in furs—a dress which I have "
+    "already adopted, for there is a great difference between walking the deck "
+    "and remaining seated motionless for hours, when no exercise prevents "
+    "the blood from actually freezing in your veins. I have no ambition to "
+    "lose my life on the post-road between St. Petersburgh and Archangel.\n"
+    "I shall depart for the latter town in a fortnight or three weeks; and my "
+    "intention is to hire a ship there, which can easily be done by paying "
+    "the insurance for the owner, and to engage as many sailors as I think "
+    "necessary among those who are accustomed to the whale-fishing. I do not "
+    "intend to sail until the month of June; and when shall I return? Ah, "
+    "dear sister, how can I answer this question? If I succeed, many, many "
+    "months, perhaps years, will pass before you and I may meet. If I fail, "
+    "you will see me again soon, or never.\n"
+    "Farewell, my dear, excellent Margaret. Heaven shower down blessings on "
+    "you, and save me, that I may again and again testify my gratitude for "
+    "all your love and kindness.\n\n"
+    "Your affectionate brother,\nR. Walton ";
+
+// Number of CHANNEL_CHUNK_LENGTH (1600) chunks required to send the
+// above text into the vc dechunker
+#define FRANKENSTEIN_VC_CHUNK_COUNT \
+    ((sizeof(frankenstein) + (CHANNEL_CHUNK_LENGTH - 1)) \
+     / CHANNEL_CHUNK_LENGTH)
+
+// 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 "                                                  "
+
+/******************************************************************************/
+/*
+  * Constructs a stream from static data
+  *
+  * The returned stream is suitable for reading.
+  */
+static struct stream *
+make_stream_from_data(const char *data, int data_len)
+{
+    struct stream *s;
+    make_stream(s);
+    init_stream(s, data_len);
+    s_push_layer(s, iso_hdr, 0);
+    out_uint8p(s, data, data_len);
+    s_mark_end(s);
+    s_pop_layer(s, iso_hdr);
+    return s;
+}
+
+/******************************************************************************/
+/*
+ * Check bad parameters passed to the dechunker functions
+ */
+START_TEST(test_vc_dechunker_bad_params)
+{
+    struct vc_dechunker *dc;
+    const char data[] = "Some stream data";
+    struct stream *s = make_stream_from_data(data, sizeof(data));
+    enum vc_dechunker_status stat;
+
+    // vc_dechunker_init
+    dc = vc_dechunker_init(NULL, 1000); // No channel name
+    ck_assert_ptr_eq(dc, NULL);
+    dc = vc_dechunker_init("test", 1); // max chunk size too small
+    ck_assert_ptr_eq(dc, NULL);
+    dc = vc_dechunker_init("test", 1000); // Should be OK
+    ck_assert_ptr_ne(dc, NULL);
+
+    // vc_dechunker_free
+    vc_dechunker_free(NULL);   // Must not crash!
+
+    // vc_dechunker_get_stream
+    vc_dechunker_get_stream(NULL);   // Must not crash!
+
+    // vc_dechunker_process_chunk
+    stat = vc_dechunker_process_chunk(NULL, s, 0, 1600); // No dechunker
+    ck_assert_int_eq(stat, E_VC_ERROR);
+    stat = vc_dechunker_process_chunk(dc, NULL, 0, 1600); // No stream
+    ck_assert_int_eq(stat, E_VC_ERROR);
+    stat = vc_dechunker_process_chunk(dc, s, 0, -1); // bad total_size
+    ck_assert_int_eq(stat, E_VC_ERROR);
+
+    free_stream(s);
+    vc_dechunker_free(dc);
+}
+
+/******************************************************************************/
+/*
+ * Check passthrough chunks (i.e. those with FIRST and LAST bits set)
+ *
+ * When the dechunker is in normal operation, these chunks are
+ * 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_vc_dechunker_passthrough)
+{
+    const char data[] = "Some data to dechunk";
+    struct stream *s = make_stream_from_data(data, sizeof(data));
+    enum vc_dechunker_status stat;
+
+    struct vc_dechunker *dc = vc_dechunker_init("test", 1000);
+    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 passthrough chunk is normally recognised immediately
+    stat = vc_dechunker_process_chunk(
+               dc, s,
+               XR_CHANNEL_FLAG_FIRST | XR_CHANNEL_FLAG_LAST,
+               s->size);
+    ck_assert_int_eq(stat, E_VC_INLINE_CHUNK);
+
+    // Check a passthrough chunk after a first chunk generates an error
+    // The total size passed for the first chunk must be bigger than the
+    // dechunker chunking_size
+    s_pop_layer(s, iso_hdr);  // Restore the stream pointer
+    stat = vc_dechunker_process_chunk(
+               dc, s,
+               XR_CHANNEL_FLAG_FIRST,
+               2000);
+    ck_assert_int_eq(stat, E_VC_IN_PROGRESS);
+
+    s_pop_layer(s, iso_hdr);
+    stat = vc_dechunker_process_chunk(
+               dc, s,
+               XR_CHANNEL_FLAG_FIRST | XR_CHANNEL_FLAG_LAST,
+               s->size);
+    ck_assert_int_eq(stat, E_VC_ERROR);
+
+    // Check a passthrough chunk is accepted after the error
+    // (i.e. the dechunker can still be used if required)
+    s_pop_layer(s, iso_hdr);
+    stat = vc_dechunker_process_chunk(
+               dc, s,
+               XR_CHANNEL_FLAG_FIRST | XR_CHANNEL_FLAG_LAST,
+               s->size);
+    ck_assert_int_eq(stat, E_VC_INLINE_CHUNK);
+
+    vc_dechunker_free(dc);
+    free_stream(s);
+}
+END_TEST
+
+
+/******************************************************************************/
+/*
+ * An intermediate chunk (neither first of last) must be rejected if we
+ * are not dechunking
+ */
+START_TEST(test_vc_dechunker_intermediate)
+{
+    const char data[] = PAD50 "Some data to dechunk";
+    struct stream *s = make_stream_from_data(data, sizeof(data));
+    enum vc_dechunker_status stat;
+
+    struct vc_dechunker *dc = vc_dechunker_init("test", sizeof(data));
+    ck_assert_ptr_ne(dc, NULL);
+
+    // Check an intermediate chunk is immediately rejected
+    stat = vc_dechunker_process_chunk(
+               dc, s,
+               0,
+               s->size + 1);
+    ck_assert_int_eq(stat, E_VC_ERROR);
+
+    vc_dechunker_free(dc);
+    free_stream(s);
+}
+END_TEST
+
+/******************************************************************************/
+/*
+ * A LAST chunk must be rejected if we are not dechunking
+ */
+START_TEST(test_vc_dechunker_last)
+{
+    const char data[] = PAD50 "Some data to dechunk";
+    struct stream *s = make_stream_from_data(data, sizeof(data));
+    enum vc_dechunker_status stat;
+
+    struct vc_dechunker *dc = vc_dechunker_init("test", sizeof(data));
+    ck_assert_ptr_ne(dc, NULL);
+
+    // Check a LAST chunk is immediately rejected
+    stat = vc_dechunker_process_chunk(
+               dc, s,
+               XR_CHANNEL_FLAG_LAST,
+               s->size + 1);
+    ck_assert_int_eq(stat, E_VC_ERROR);
+
+    vc_dechunker_free(dc);
+    free_stream(s);
+}
+END_TEST
+
+/******************************************************************************/
+/**
+ * Two consecutive FIRST chunks are not allowed
+ */
+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));
+    enum vc_dechunker_status stat;
+
+    struct vc_dechunker *dc = vc_dechunker_init("test", sizeof(data));
+    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 = vc_dechunker_process_chunk(
+               dc, s,
+               XR_CHANNEL_FLAG_FIRST,
+               s->size + 1);
+    ck_assert_int_eq(stat, E_VC_IN_PROGRESS);
+
+    // ... and another FIRST chunk is an error
+    s_pop_layer(s, iso_hdr);
+    stat = vc_dechunker_process_chunk(
+               dc, s,
+               XR_CHANNEL_FLAG_FIRST,
+               s->size + 1);
+    ck_assert_int_eq(stat, E_VC_ERROR);
+
+    vc_dechunker_free(dc);
+    free_stream(s);
+}
+END_TEST
+
+
+/******************************************************************************/
+/**
+ * Checks that a FIRST chunk cannot be the total_size. This follows from
+ * [MS-RDPBCGR] 3.1.5.2.1:
+ *
+ * > If the total size of the virtual channel data is larger than
+ * > the chunk size, then each chunk MUST be sent in a separate Virtual
+ * > Channel PDU.
+*
+*  > Virtual channel data that fits in a single Virtual Channel PDU MUST
+*  > specify both flags
+ */
+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));
+    enum vc_dechunker_status stat;
+
+    struct vc_dechunker *dc = vc_dechunker_init("test", sizeof(data));
+    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 the same size as the total size is rejected
+    stat = vc_dechunker_process_chunk(
+               dc, s,
+               XR_CHANNEL_FLAG_FIRST,
+               s->size);
+    ck_assert_int_eq(stat, E_VC_ERROR);
+
+    // Check a FIRST chunk bigger than the total size is rejected
+    // [MS-RDPBCGR] 3.1.5.2.1
+    vc_dechunker_free(dc);
+    dc = vc_dechunker_init("test", sizeof(data));
+    ck_assert_ptr_ne(dc, NULL);
+    s_pop_layer(s, iso_hdr);
+    stat = vc_dechunker_process_chunk(
+               dc, s,
+               XR_CHANNEL_FLAG_FIRST,
+               s->size - 1);
+    ck_assert_int_eq(stat, E_VC_ERROR);
+
+    vc_dechunker_free(dc);
+    free_stream(s);
+}
+END_TEST
+
+/******************************************************************************/
+// Returns a stream with some random data, then a chunk of
+// 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_vc_bigtest_chunk(unsigned int chunk_num, int *flags)
+{
+    struct stream *s = NULL;
+
+    if (chunk_num < FRANKENSTEIN_VC_CHUNK_COUNT)
+    {
+        int chunk_size;
+        // Work out the size of this chunk
+        if (chunk_num == (FRANKENSTEIN_VC_CHUNK_COUNT - 1))
+        {
+            chunk_size = sizeof(frankenstein) % CHANNEL_CHUNK_LENGTH;
+            if (chunk_size == 0)
+            {
+                chunk_size = CHANNEL_CHUNK_LENGTH;
+            }
+        }
+        else
+        {
+            chunk_size = CHANNEL_CHUNK_LENGTH;
+        }
+
+        // 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 * CHANNEL_CHUNK_LENGTH],
+                   chunk_size);
+
+        // Get the stream ready for reading from the Frankenstein text
+        s_mark_end(s);
+        s_pop_layer(s, iso_hdr);
+
+        // Sort out the flags
+        *flags =
+            (chunk_num == 0) ? XR_CHANNEL_FLAG_FIRST :
+            (chunk_num == (FRANKENSTEIN_VC_CHUNK_COUNT - 1)) ? XR_CHANNEL_FLAG_LAST :
+            0;
+    }
+
+    return s;
+}
+
+/******************************************************************************/
+/*
+ * Streams a lot of data through the dechunker and checks it's all
+ * assembled correctly at the end
+ */
+START_TEST(test_vc_dechunker_big_test)
+{
+    enum vc_dechunker_status stat;
+    struct vc_dechunker *dc = vc_dechunker_init("test", CHANNEL_CHUNK_LENGTH);
+    ck_assert_ptr_ne(dc, NULL);
+    struct stream *s;
+
+    int i;
+    for (i = 0 ; i < FRANKENSTEIN_VC_CHUNK_COUNT; ++i)
+    {
+        int flags;
+        s = make_vc_bigtest_chunk(i, &flags);
+        stat = vc_dechunker_process_chunk(
+                   dc, s,
+                   flags,
+                   sizeof(frankenstein));
+        if (i < FRANKENSTEIN_VC_CHUNK_COUNT - 1)
+        {
+            ck_assert_int_eq(stat, E_VC_IN_PROGRESS);
+        }
+        else
+        {
+            ck_assert_int_eq(stat, E_VC_READY);
+        }
+        free_stream(s);
+    }
+
+    // Check we have a result
+    s = vc_dechunker_get_stream(dc);
+    ck_assert_ptr_ne(s, NULL);
+
+    // 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);
+    vc_dechunker_free(dc);
+}
+
+/******************************************************************************/
+// Like test_vc_dechunker_big_test, but the last chunk is oversized
+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);
+    ck_assert_ptr_ne(dc, NULL);
+    struct stream *s;
+
+    int i;
+    for (i = 0 ; i < FRANKENSTEIN_VC_CHUNK_COUNT; ++i)
+    {
+        int flags;
+        s = make_vc_bigtest_chunk(i, &flags);
+        if (i == (FRANKENSTEIN_VC_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 = vc_dechunker_process_chunk(
+                   dc, s,
+                   flags,
+                   sizeof(frankenstein));
+        if (i < FRANKENSTEIN_VC_CHUNK_COUNT - 1)
+        {
+            ck_assert_int_eq(stat, E_VC_IN_PROGRESS);
+        }
+        else
+        {
+            ck_assert_int_eq(stat, E_VC_ERROR);
+        }
+        free_stream(s);
+    }
+
+    vc_dechunker_free(dc);
+}
+
+/******************************************************************************/
+// Like test_vc_dechunker_big_test, but the last chunk is undersized
+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);
+    ck_assert_ptr_ne(dc, NULL);
+    struct stream *s;
+
+    int i;
+    for (i = 0 ; i < FRANKENSTEIN_VC_CHUNK_COUNT; ++i)
+    {
+        int flags;
+        s = make_vc_bigtest_chunk(i, &flags);
+        if (i == (FRANKENSTEIN_VC_CHUNK_COUNT - 1))
+        {
+            // Skip a byte in the stream, so there is one
+            // less byte than expected
+            in_uint8s(s, 1);
+        }
+        stat = vc_dechunker_process_chunk(
+                   dc, s,
+                   flags,
+                   sizeof(frankenstein));
+        if (i < FRANKENSTEIN_VC_CHUNK_COUNT - 1)
+        {
+            ck_assert_int_eq(stat, E_VC_IN_PROGRESS);
+        }
+        else
+        {
+            ck_assert_int_eq(stat, E_VC_ERROR);
+        }
+        free_stream(s);
+    }
+
+    vc_dechunker_free(dc);
+}
+
+/******************************************************************************/
+
+/******************************************************************************/
+/*
+ * 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)
+{
+    Suite *s;
+    TCase *rc_dechunker;
+
+    s = suite_create("dechunker");
+
+    rc_dechunker = tcase_create("vc_dechunker");
+    suite_add_tcase(s, rc_dechunker);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_bad_params);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_passthrough);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_intermediate);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_last);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_first_first);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_chunk_overflow);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_big_test);
+    tcase_add_test(rc_dechunker, test_vc_dechunker_big_test_oversize_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);
 }