Per-channel stderr buffering and flow control

Move extDataBuffer from WOLFSSH to WOLFSSH_CHANNEL and give stderr real
back-pressure: charge the window on receipt, credit on read, never lose
the credit.

- Buffer stderr per channel; accumulate across packets instead of
  overwriting unread data.
- Charge the window on receipt, replenish on read; reject data larger
  than the advertised window.
- Route all credit through ChannelCreditWindow(): parks credit that
  cannot go out (mid-rekey, failed send), suppresses zero-byte adjusts.
- Add wolfSSH_Channel{Id,}ReadExt / SendExt; the stream_/extended_data_
  pair stays first-channel-only.
- Drain SCP stderr in a loop; document the mandatory-drain contract.

Issue: F-864

Author: ejohnstown

src/internal.c

@@ -1533,8 +1533,7 @@ WOLFSSH* SshInit(WOLFSSH* ssh, WOLFSSH_CTX* ctx)
     ssh->keyingCompletionCtx = (void*)ssh;
 
     if (BufferInit(&ssh->inputBuffer, 0, ctx->heap) != WS_SUCCESS  ||
-        BufferInit(&ssh->outputBuffer, 0, ctx->heap) != WS_SUCCESS ||
-        BufferInit(&ssh->extDataBuffer, 0, ctx->heap) != WS_SUCCESS) {
+        BufferInit(&ssh->outputBuffer, 0, ctx->heap) != WS_SUCCESS) {
 
         wolfSSH_free(ssh);
         ssh = NULL;
@@ -1553,7 +1552,6 @@ void SshResourceFree(WOLFSSH* ssh, void* heap)
 
     ShrinkBuffer(&ssh->inputBuffer, 1);
     ShrinkBuffer(&ssh->outputBuffer, 1);
-    ShrinkBuffer(&ssh->extDataBuffer, 1);
     WS_FORCEZERO(ssh->k, ssh->kSz);
     HandshakeInfoFree(ssh->handshake, heap);
     WS_FORCEZERO(&ssh->keys, sizeof(Keys));
@@ -3523,6 +3521,14 @@ WOLFSSH_CHANNEL* ChannelNew(WOLFSSH* ssh, byte channelType,
                 newChannel->inputBuffer.buffer = buffer;
                 newChannel->inputBuffer.bufferSz = initialWindowSz;
                 newChannel->inputBuffer.dynamicFlag = 1;
+                if (BufferInit(&newChannel->extDataBuffer, 0, heap)
+                        != WS_SUCCESS) {
+                    WLOG(WS_LOG_DEBUG,
+                         "Unable to init new channel's ext data buffer");
+                    WFREE(buffer, heap, DYNTYPE_BUFFER);
+                    WFREE(newChannel, heap, DYNTYPE_CHANNEL);
+                    newChannel = NULL;
+                }
             }
             else {
                 WLOG(WS_LOG_DEBUG, "Unable to allocate new channel's buffer");
@@ -3554,6 +3560,14 @@ void ChannelDelete(WOLFSSH_CHANNEL* channel, void* heap)
     #endif /* WOLFSSH_FWD */
         WFREE(channel->inputBuffer.buffer,
               channel->inputBuffer.heap, DYNTYPE_BUFFER);
+        if (channel->extDataBuffer.length > channel->extDataBuffer.idx) {
+            /* The app never drained it and the channel is going away. */
+            WLOG(WS_LOG_INFO,
+                 "Discarding %u bytes of unread extended data on channel %u",
+                 channel->extDataBuffer.length - channel->extDataBuffer.idx,
+                 channel->channel);
+        }
+        ShrinkBuffer(&channel->extDataBuffer, 1);
         if (channel->command)
             WFREE(channel->command, heap, DYNTYPE_STRING);
         WFREE(channel, heap, DYNTYPE_CHANNEL);
@@ -7176,6 +7190,84 @@ static int DoKexDhReply(WOLFSSH* ssh, byte* buf, word32 len, word32* idx)
 }
 
 
+/* Returns amount bytes of receive-window credit to the peer, folding in credit
+ * already parked on the channel. Credit that cannot reach the transport is
+ * parked, not dropped: no WINDOW_ADJUST may be sent mid-rekey (RFC 4253 section
+ * 7.1), an unbundled packet queued nothing, and a socket error can discard what
+ * was bundled. Credit that reached the output buffer counts as delivered. */
+int ChannelCreditWindow(WOLFSSH* ssh, WOLFSSH_CHANNEL* channel, word32 amount)
+{
+    word32 total;
+    int ret;
+    byte bundled = 0;
+
+    if (ssh == NULL || channel == NULL)
+        return WS_BAD_ARGUMENT;
+
+    if (amount > UINT32_MAX - channel->pendingWindowAdjust) {
+        WLOG(WS_LOG_ERROR, "pending window adjust would overflow");
+        return WS_OVERFLOW_E;
+    }
+
+    total = channel->pendingWindowAdjust + amount;
+    if (total == 0)
+        return WS_SUCCESS;
+
+    if (ssh->isKeying) {
+        channel->pendingWindowAdjust = total;
+        return WS_SUCCESS;
+    }
+
+    channel->pendingWindowAdjust = 0;
+    ret = SendChannelWindowAdjust(ssh, channel->channel, total, &bundled);
+
+    /* Keep the credit owed only if it never reached the transport. Once bundled
+     * it sits in the output buffer for a later flush, so re-parking after a
+     * short write (WS_WANT_WRITE) would send it twice and inflate the peer's
+     * window. WS_SOCKET_ERROR_E is the exception: wolfSSH_SendPacket() can
+     * discard what it bundled, so it stays owed. */
+    if (!bundled || ret == WS_SOCKET_ERROR_E) {
+        channel->pendingWindowAdjust = total;
+        WLOG(WS_LOG_ERROR,
+             "ChannelCreditWindow: window adjust send failed (%d) for "
+             "channel %u; %u bytes of credit parked for retry",
+             ret, channel->channel, total);
+    }
+
+    return ret;
+}
+
+
+/* Flush receive-window credit parked on the channels. Called when keying
+ * completes. */
+static int SendPendingChannelWindowAdjust(WOLFSSH* ssh)
+{
+    WOLFSSH_CHANNEL* cur;
+    int savedError;
+
+    if (ssh == NULL)
+        return WS_BAD_ARGUMENT;
+
+    /* Still mid-KEX: leave the credit parked until both sides finish. */
+    if (ssh->isKeying)
+        return WS_SUCCESS;
+
+    /* Save and restore ssh->error: this flush is incidental to the keying flow,
+     * and the caller (DoNewKeys/SendNewKeys) reports its own result. A bundled
+     * adjust that cannot flush in non-blocking mode sets WS_WANT_WRITE but still
+     * goes out on the next flush. */
+    savedError = ssh->error;
+
+    for (cur = ssh->channelList; cur != NULL; cur = cur->next) {
+        if (cur->pendingWindowAdjust != 0)
+            ChannelCreditWindow(ssh, cur, 0);
+    }
+
+    ssh->error = savedError;
+    return WS_SUCCESS;
+}
+
+
 static int DoNewKeys(WOLFSSH* ssh, byte* buf, word32 len, word32* idx)
 {
     int ret = WS_SUCCESS;
@@ -7218,6 +7310,9 @@ static int DoNewKeys(WOLFSSH* ssh, byte* buf, word32 len, word32* idx)
         ssh->isKeying &= ~WOLFSSH_PEER_IS_KEYING;
         HandshakeInfoFree(ssh->handshake, ssh->ctx->heap);
         ssh->handshake = NULL;
+        /* If this was the last keying flag to clear, flush any window credit
+         * that was deferred during the rekey (RFC 4253 section 7.1). */
+        SendPendingChannelWindowAdjust(ssh);
         WLOG(WS_LOG_DEBUG, "Keying completed");
         if (ssh->ctx->keyingCompletionCb)
             ssh->ctx->keyingCompletionCb(ssh->keyingCompletionCtx);
@@ -11153,25 +11248,45 @@ static int DoChannelData(WOLFSSH* ssh,
 }
 
 
-/* deletes current buffer and updates it
- * return WS_SUCCESS on success */
-static int PutBuffer(WOLFSSH_BUFFER* buf, byte* data, word32 dataSz)
+/* Appends data, preserving unread bytes. Already-read data (below idx) is
+ * compacted away and the new data is appended after the unread region. */
+static int AppendBuffer(WOLFSSH_BUFFER* buf, byte* data, word32 dataSz)
 {
     int ret;
+    word32 usedSz, needSz, growSz;
 
-    /* reset "used" section of buffer back to 0 */
-    buf->length = 0;
-    buf->idx    = 0;
+    if (buf == NULL || data == NULL)
+        return WS_BAD_ARGUMENT;
 
-    if (dataSz > buf->bufferSz) {
-        if ((ret = GrowBuffer(buf, dataSz)) != WS_SUCCESS) {
-            return ret;
+    usedSz = buf->length - buf->idx;
+    if (dataSz > UINT32_MAX - usedSz)
+        return WS_OVERFLOW_E;
+    needSz = usedSz + dataSz;
+
+    /* GrowBuffer's size is the room wanted past the unread bytes; it reallocates
+     * when sz + usedSz exceeds bufferSz. Over-ask only when the allocation
+     * cannot hold the data, targeting double the capacity so trickled blobs cost
+     * amortized O(1) per byte instead of reallocating on every append. */
+    growSz = dataSz;
+    if (needSz > buf->bufferSz) {
+        word32 targetSz = needSz;
+
+        if (buf->bufferSz <= (UINT32_MAX / 2) &&
+                (buf->bufferSz * 2) > targetSz) {
+            targetSz = buf->bufferSz * 2;
         }
+        growSz = targetSz - usedSz;
     }
-    WMEMCPY(buf->buffer, data, dataSz);
-    buf->length = dataSz;
 
-    return WS_SUCCESS;
+    /* GrowBuffer compacts the consumed prefix (bytes below idx) down to the
+     * front. On return idx is 0 and length is the count of unread bytes. */
+    ret = GrowBuffer(buf, growSz);
+    if (ret == WS_SUCCESS) {
+        WMEMCPY(buf->buffer + buf->length, data, dataSz);
+        buf->length += dataSz;
+    }
+
+    return ret;
 }
 
 
@@ -11199,24 +11314,26 @@ static int DoChannelExtendedData(WOLFSSH* ssh,
             ret = WS_INVALID_CHANID;
         else if (dataSz > channel->maxPacketSz)
             ret = WS_RECV_OVERFLOW_E;
+        else if (dataSz > channel->windowSz)
+            /* Peer sent more than the window we advertised. */
+            ret = WS_RECV_OVERFLOW_E;
         else if (dataTypeCode == CHANNEL_EXTENDED_DATA_STDERR) {
-            ret = PutBuffer(&ssh->extDataBuffer, buf + begin, dataSz);
+            ret = AppendBuffer(&channel->extDataBuffer, buf + begin, dataSz);
             #ifdef DEBUG_WOLFSSH
             DumpOctetString(buf + begin, dataSz);
             #endif
-            if (ret == WS_SUCCESS)
-                ret = SendChannelWindowAdjust(ssh, channel->channel, dataSz);
             if (ret == WS_SUCCESS) {
+                channel->windowSz -= dataSz;
                 ssh->lastRxId = channelId;
                 ret = WS_EXTDATA;
             }
         }
         else {
             WLOG(WS_LOG_INFO, "Ignoring unknown extended data type %u",
                     dataTypeCode);
-            ret = SendChannelWindowAdjust(ssh, channel->channel, dataSz);
-            if (ret == WS_SUCCESS)
-                ssh->lastRxId = channelId;
+            (void)ChannelCreditWindow(ssh, channel, dataSz);
+            ssh->lastRxId = channelId;
+            ret = WS_SUCCESS;
         }
         *idx = begin + dataSz;
     }
@@ -15033,6 +15150,10 @@ int SendNewKeys(WOLFSSH* ssh)
 
         /* Clear self is keying flag */
         ssh->isKeying &= ~WOLFSSH_SELF_IS_KEYING;
+
+        /* If this was the last keying flag to clear, flush any window credit
+         * that was deferred during the rekey (RFC 4253 section 7.1). */
+        SendPendingChannelWindowAdjust(ssh);
     }
 
     WLOG(WS_LOG_DEBUG, "Leaving SendNewKeys(), ret = %d", ret);
@@ -19122,8 +19243,11 @@ int SendChannelExtendedData(WOLFSSH* ssh, word32 channelId,
 }
 
 
+/* Sends a WINDOW_ADJUST for bytesToAdd. When bundled is not NULL it is set to 1
+ * once the packet is in the output buffer, letting a caller tell an adjust that
+ * was never queued from one that was queued but whose flush did not finish. */
 int SendChannelWindowAdjust(WOLFSSH* ssh, word32 channelId,
-                            word32 bytesToAdd)
+                            word32 bytesToAdd, byte* bundled)
 {
     byte* output;
     word32 idx;
@@ -19164,8 +19288,14 @@ int SendChannelWindowAdjust(WOLFSSH* ssh, word32 channelId,
         ret = BundlePacket(ssh);
     }
 
-    if (ret == WS_SUCCESS)
+    if (ret == WS_SUCCESS) {
+        /* Queued. From here the send either delivers it or reports the error
+         * that discarded it; either way it was not lost before reaching the
+         * transport. */
+        if (bundled != NULL)
+            *bundled = 1;
         ret = wolfSSH_SendPacket(ssh);
+    }
 
     WLOG(WS_LOG_DEBUG, "Leaving SendChannelWindowAdjust(), ret = %d", ret);
     return ret;
@@ -20164,6 +20294,11 @@ int wolfSSH_TestDoChannelWindowAdjust(WOLFSSH* ssh, byte* buf, word32 len,
     return DoChannelWindowAdjust(ssh, buf, len, idx);
 }
 
+int wolfSSH_TestSendPendingChannelWindowAdjust(WOLFSSH* ssh)
+{
+    return SendPendingChannelWindowAdjust(ssh);
+}
+
 int wolfSSH_TestDoUserAuthRequest(WOLFSSH* ssh, byte* buf, word32 len,
         word32* idx)
 {