ssh.c: report the byte count from a channel read that defers its credit

- wolfSSH_stream_read() advances inputBuffer->idx before crediting the
  window, and it and _ChannelRead() return the bytes copied with a
  non-success adjust left in ssh->error.
- _ChannelRead() takes the WOLFSSH from channel->ssh, rejects an idx
  past inputBuffer->length, restores the entry ssh->error on a clean
  credit, and retires a stale WS_WANT_WRITE only when its own credit
  went out.
- wolfSSH_SFTP_Close() checks NoticeError() only on a failed send.
- The src/ssh.c block comments, and new wolfssh/ssh.h notes above
  wolfSSH_stream_read(), wolfSSH_ChannelRead() and
  wolfSSH_ChannelIdRead(), state the window-adjust and ssh->error
  contract.
- tests/unit.c adds test_stream_read_deferredWindowAdjust() and
  test_ChannelIdRead_deferredWindowAdjust(): an adjust that defers,
  then fails, then succeeds, plus a read with nothing buffered against
  a seeded WS_WANT_WRITE.

Author: yosuke-wolfssl

src/ssh.c

@@ -1300,9 +1300,9 @@ static int _ChannelReadExt(WOLFSSH_CHANNEL* channel, byte* buf, word32 bufSz);
  * the SSH connection. This function handles low level operations in addition to
  * the read, such as window adjustment and high water checking.
  *
- * In non blocking mode use the function wolfSSH_get_error(ssh) to check for
- * WS_WANT_READ / WS_WANT_WRITE after a fail case was hit with
- * wolfSSH_stream_read().
+ * In non blocking mode check wolfSSH_get_error(ssh) after the read: it holds
+ * WS_WANT_READ / WS_WANT_WRITE for a fail case, and for a success the status
+ * of a window adjust that could not be sent.
  *
  * Returns the number of bytes read on success, negative values on fail
  */
@@ -1394,11 +1394,17 @@ int wolfSSH_stream_read(WOLFSSH* ssh, byte* buf, word32 bufSz)
             ret = WS_BUFFER_E;
         else {
             WMEMCPY(buf, inputBuffer->buffer + inputBuffer->idx, n);
+            inputBuffer->idx += n;
             ret = _UpdateChannelWindow(ssh->channelList);
-            if (ret == WS_SUCCESS) {
-                inputBuffer->idx += n;
-                ret = n;
+            if (ret != WS_SUCCESS) {
+                ssh->error = ret;
+                if (ret != WS_WANT_WRITE) {
+                    WLOG(WS_LOG_ERROR,
+                         "wolfSSH_stream_read: window adjust send failed "
+                         "(%d); read still succeeded", ret);
+                }
             }
+            ret = n;
         }
     }
 
@@ -4000,24 +4006,59 @@ static int _UpdateChannelWindow(WOLFSSH_CHANNEL* channel)
 }
 
 
+/* Drains buffered channel data and credits the window for the bytes taken.
+ * Reports the bytes copied; an adjust that could not go out lands in
+ * ssh->error. */
 static int _ChannelRead(WOLFSSH_CHANNEL* channel, byte* buf, word32 bufSz)
 {
     WOLFSSH_BUFFER* inputBuffer;
+    WOLFSSH* ssh;
+    word32 creditedSz;
     int updateResult = WS_SUCCESS;
+    int savedError;
 
     if (channel == NULL || buf == NULL || bufSz == 0)
         return WS_BAD_ARGUMENT;
 
+    ssh = channel->ssh;
     inputBuffer = &channel->inputBuffer;
+
+    if (inputBuffer->idx > inputBuffer->length) {
+        WLOG(WS_LOG_ERROR, "Bad internal state for buffer index");
+        return WS_INVALID_STATE_E;
+    }
+
     bufSz = min(bufSz, inputBuffer->length - inputBuffer->idx);
     WMEMCPY(buf, inputBuffer->buffer + inputBuffer->idx, bufSz);
     inputBuffer->idx += bufSz;
 
+    /* Unguarded by bufSz: also compacts, and carries credit left behind. */
+    savedError = ssh->error;
+    creditedSz = inputBuffer->idx;
     updateResult = _UpdateChannelWindow(channel);
-    if (updateResult == WS_SUCCESS)
-        updateResult = bufSz;
+    if (updateResult == WS_SUCCESS) {
+        /* Clear the old WS_WANT_WRITE only if this read sent an adjust of
+         * its own and the output buffer is now empty. */
+        if (savedError == WS_WANT_WRITE && creditedSz != 0
+                && inputBuffer->idx == 0 && ssh->outputBuffer.length == 0) {
+            ssh->error = WS_SUCCESS;
+        }
+        else {
+            ssh->error = savedError;
+        }
+    }
+    else {
+        /* SendPacket() records only WS_WANT_WRITE, so a hard failure has to
+         * be recorded here; rewriting WS_WANT_WRITE is deliberate. */
+        ssh->error = updateResult;
+        if (updateResult != WS_WANT_WRITE) {
+            WLOG(WS_LOG_ERROR,
+                 "_ChannelRead: window adjust send failed (%d); read still "
+                 "succeeded", updateResult);
+        }
+    }
 
-    return updateResult;
+    return (int)bufSz;
 }
 
 

src/wolfsftp.c

@@ -8914,7 +8914,7 @@ int wolfSSH_SFTP_Close(WOLFSSH* ssh, byte* handle, word32 handleSz)
             case STATE_CLOSE_SEND:
                 WLOG(WS_LOG_SFTP, "SFTP CLOSE STATE: SEND");
                 ret = SendPacketType(ssh, WOLFSSH_FTP_CLOSE, handle, handleSz);
-                if (NoticeError(ssh)) {
+                if (ret != WS_SUCCESS && NoticeError(ssh)) {
                     return WS_FATAL_ERROR;
                 }
 

tests/unit.c

@@ -6488,6 +6488,220 @@ static int test_SendChannelData_zeroPeerMaxPacket(void)
     return result;
 }
 
+#ifndef NO_WOLFSSH_SERVER
+
+/* wolfSSH_stream_read() counterpart of test_ChannelExtDataCreditWantWrite():
+ * a deferred credit must not cost the caller the bytes already consumed, and
+ * an adjust that fails outright must still surface on wolfSSH_get_error(). */
+static int test_stream_read_deferredWindowAdjust(void)
+{
+    WOLFSSH_CTX*     ctx = NULL;
+    WOLFSSH*         ssh = NULL;
+    WOLFSSH_CHANNEL* ch  = NULL;
+    int              result = 0;
+    int              ret;
+    byte             in[64];
+    byte             out[64];
+    word32           i;
+
+    for (i = 0; i < (word32)sizeof(in); i++) {
+        in[i] = (byte)i;
+    }
+
+    ctx = wolfSSH_CTX_new(WOLFSSH_ENDPOINT_SERVER, NULL);
+    if (ctx == NULL)
+        return -6980;
+    wolfSSH_SetIOSend(ctx, WantWriteIoSend);
+
+    ssh = wolfSSH_new(ctx);
+    if (ssh == NULL) { result = -6981; goto done; }
+    /* Allow MSGID_CHANNEL_WINDOW_ADJUST on this bare session. */
+    ssh->acceptState = ACCEPT_SERVER_USERAUTH_SENT;
+
+    /* A window the size of the payload, so draining it in one read leaves
+     * windowSz at zero and _UpdateChannelWindow() has to credit. */
+    ch = ChannelNew(ssh, ID_CHANTYPE_SESSION,
+                    (word32)sizeof(in), DEFAULT_MAX_PACKET_SZ);
+    if (ch == NULL) { result = -6982; goto done; }
+    if (ChannelAppend(ssh, ch) != WS_SUCCESS) {
+        ChannelDelete(ch, ssh->ctx->heap);
+        result = -6983;
+        goto done;
+    }
+    ch->openConfirmed = 1;
+
+    if (wolfSSH_TestChannelPutData(ch, in, (word32)sizeof(in)) != WS_SUCCESS) {
+        result = -6984; goto done;
+    }
+    if (ch->windowSz != 0) { result = -6985; goto done; }
+
+    /* Every byte is reported, and they are the bytes that were put. */
+    ret = wolfSSH_stream_read(ssh, out, (word32)sizeof(out));
+    if (ret != (int)sizeof(in)) { result = -6990; goto done; }
+    if (WMEMCMP(out, in, sizeof(in)) != 0) { result = -6991; goto done; }
+
+    /* The deferral is observable, the window is credited locally, and the
+     * bytes are consumed rather than left for a re-read. */
+    if (ssh->error != WS_WANT_WRITE) { result = -6992; goto done; }
+    if (ch->windowSz != (word32)sizeof(in)) { result = -6993; goto done; }
+    if (ch->inputBuffer.length - ch->inputBuffer.idx != 0) {
+        result = -6994; goto done;
+    }
+
+    /* A peer that reset rather than blocked. wolfSSH_SendPacket() records only
+     * WS_WANT_WRITE, so the read path has to record a hard failure itself. */
+    wolfSSH_SetIOSend(ctx, FailIoSend);
+
+    if (wolfSSH_TestChannelPutData(ch, in, (word32)sizeof(in)) != WS_SUCCESS) {
+        result = -6995; goto done;
+    }
+    if (ch->windowSz != 0) { result = -6996; goto done; }
+
+    ret = wolfSSH_stream_read(ssh, out, (word32)sizeof(out));
+    if (ret != (int)sizeof(in)) { result = -6997; goto done; }
+    if (wolfSSH_get_error(ssh) != WS_SOCKET_ERROR_E) {
+        result = -6998; goto done;
+    }
+    /* The send discarded what it bundled, so the credit stays owed. */
+    if (ch->pendingWindowAdjust != (word32)sizeof(in)) {
+        result = -6999; goto done;
+    }
+
+    /* Credit is charged for the bytes this read took, so a read that drains
+     * the window puts an adjust on the wire. */
+    wolfSSH_SetIOSend(ctx, CountIoSend);
+    s_extSendCount = 0;
+
+    if (wolfSSH_TestChannelPutData(ch, in, (word32)sizeof(in)) != WS_SUCCESS) {
+        result = -6970; goto done;
+    }
+
+    ret = wolfSSH_stream_read(ssh, out, (word32)sizeof(out));
+    if (ret != (int)sizeof(in)) { result = -6971; goto done; }
+    if (s_extSendCount != 1) { result = -6972; goto done; }
+    if (wolfSSH_get_error(ssh) != WS_SUCCESS) { result = -6973; goto done; }
+    /* The owed credit from the failed send rode along with this one. */
+    if (ch->pendingWindowAdjust != 0) { result = -6974; goto done; }
+    if (ch->windowSz != (word32)sizeof(in)) { result = -6975; goto done; }
+
+done:
+    wolfSSH_free(ssh);
+    wolfSSH_CTX_free(ctx);
+    return result;
+}
+
+/* wolfSSH_ChannelIdRead() counterpart of
+ * test_stream_read_deferredWindowAdjust(): callers break out on a non-positive
+ * read, and this entry point has to retire the owed-flush status itself. */
+static int test_ChannelIdRead_deferredWindowAdjust(void)
+{
+    WOLFSSH_CTX*     ctx = NULL;
+    WOLFSSH*         ssh = NULL;
+    WOLFSSH_CHANNEL* ch  = NULL;
+    int              result = 0;
+    int              ret;
+    byte             in[64];
+    byte             out[64];
+    word32           i;
+
+    for (i = 0; i < (word32)sizeof(in); i++) {
+        in[i] = (byte)i;
+    }
+
+    ctx = wolfSSH_CTX_new(WOLFSSH_ENDPOINT_SERVER, NULL);
+    if (ctx == NULL)
+        return -7010;
+    wolfSSH_SetIOSend(ctx, WantWriteIoSend);
+
+    ssh = wolfSSH_new(ctx);
+    if (ssh == NULL) { result = -7011; goto done; }
+    /* Allow MSGID_CHANNEL_WINDOW_ADJUST on this bare session. */
+    ssh->acceptState = ACCEPT_SERVER_USERAUTH_SENT;
+
+    /* A window the size of the payload, so draining it in one read leaves
+     * windowSz at zero and _UpdateChannelWindow() has to credit. */
+    ch = ChannelNew(ssh, ID_CHANTYPE_SESSION,
+                    (word32)sizeof(in), DEFAULT_MAX_PACKET_SZ);
+    if (ch == NULL) { result = -7012; goto done; }
+    if (ChannelAppend(ssh, ch) != WS_SUCCESS) {
+        ChannelDelete(ch, ssh->ctx->heap);
+        result = -7013;
+        goto done;
+    }
+    ch->openConfirmed = 1;
+
+    if (wolfSSH_TestChannelPutData(ch, in, (word32)sizeof(in)) != WS_SUCCESS) {
+        result = -7014; goto done;
+    }
+    if (ch->windowSz != 0) { result = -7015; goto done; }
+
+    /* Unlike wolfSSH_stream_read(), this entry point does not clear the error,
+     * so seed it: the assert below has to prove the read recorded it. */
+    ssh->error = WS_SUCCESS;
+
+    /* Every byte is reported, and they are the bytes that were put. */
+    ret = wolfSSH_ChannelIdRead(ssh, ch->channel, out, (word32)sizeof(out));
+    if (ret != (int)sizeof(in)) { result = -7016; goto done; }
+    if (WMEMCMP(out, in, sizeof(in)) != 0) { result = -7017; goto done; }
+
+    /* The deferral is observable, the window is credited locally, and the
+     * bytes are consumed rather than left for a re-read. */
+    if (ssh->error != WS_WANT_WRITE) { result = -7018; goto done; }
+    if (ch->windowSz != (word32)sizeof(in)) { result = -7019; goto done; }
+    if (ch->inputBuffer.length - ch->inputBuffer.idx != 0) {
+        result = -7020; goto done;
+    }
+
+    /* A peer that reset rather than blocked. wolfSSH_SendPacket() records only
+     * WS_WANT_WRITE, so the read path has to record a hard failure itself. */
+    wolfSSH_SetIOSend(ctx, FailIoSend);
+
+    if (wolfSSH_TestChannelPutData(ch, in, (word32)sizeof(in)) != WS_SUCCESS) {
+        result = -7021; goto done;
+    }
+    if (ch->windowSz != 0) { result = -7022; goto done; }
+
+    ret = wolfSSH_ChannelIdRead(ssh, ch->channel, out, (word32)sizeof(out));
+    if (ret != (int)sizeof(in)) { result = -7023; goto done; }
+    if (ssh->error != WS_SOCKET_ERROR_E) { result = -7024; goto done; }
+    /* The send discarded what it bundled, so the credit stays owed. */
+    if (ch->pendingWindowAdjust != (word32)sizeof(in)) {
+        result = -7025; goto done;
+    }
+
+    /* This entry point never resets ssh->error, so a credit that does go out
+     * has to retire the owed-flush status itself. */
+    wolfSSH_SetIOSend(ctx, DiscardIoSend);
+    ssh->error = WS_WANT_WRITE;
+
+    if (wolfSSH_TestChannelPutData(ch, in, (word32)sizeof(in)) != WS_SUCCESS) {
+        result = -7026; goto done;
+    }
+
+    ret = wolfSSH_ChannelIdRead(ssh, ch->channel, out, (word32)sizeof(out));
+    if (ret != (int)sizeof(in)) { result = -7027; goto done; }
+    if (ssh->outputBuffer.length != 0) { result = -7028; goto done; }
+    if (wolfSSH_get_error(ssh) != WS_SUCCESS) { result = -7029; goto done; }
+    /* Both the parked credit and the new one reached the peer. */
+    if (ch->pendingWindowAdjust != 0) { result = -7030; goto done; }
+
+    /* A read with nothing buffered sends no credit, so it has no standing to
+     * retire a WS_WANT_WRITE some other sender is still owed. */
+    ssh->error = WS_WANT_WRITE;
+
+    ret = wolfSSH_ChannelIdRead(ssh, ch->channel, out, (word32)sizeof(out));
+    if (ret != 0) { result = -7031; goto done; }
+    if (ssh->outputBuffer.length != 0) { result = -7032; goto done; }
+    if (wolfSSH_get_error(ssh) != WS_WANT_WRITE) { result = -7033; goto done; }
+
+done:
+    wolfSSH_free(ssh);
+    wolfSSH_CTX_free(ctx);
+    return result;
+}
+
+#endif /* NO_WOLFSSH_SERVER */
+
 /* BuildNameList() returns a C string. On an empty id list it must still
  * terminate the buffer: SendKexInit() measures the result with WSTRLEN
  * through AlgoListSz() and copies that many bytes into the KEXINIT. */
@@ -17194,6 +17408,18 @@ int wolfSSH_UnitTest(int argc, char** argv)
            (unitResult == 0 ? "SUCCESS" : "FAILED"));
     testResult = testResult || unitResult;
 
+#ifndef NO_WOLFSSH_SERVER
+    unitResult = test_stream_read_deferredWindowAdjust();
+    printf("stream_read_deferredWindowAdjust: %s\n",
+           (unitResult == 0 ? "SUCCESS" : "FAILED"));
+    testResult = testResult || unitResult;
+
+    unitResult = test_ChannelIdRead_deferredWindowAdjust();
+    printf("ChannelIdRead_deferredWindowAdjust: %s\n",
+           (unitResult == 0 ? "SUCCESS" : "FAILED"));
+    testResult = testResult || unitResult;
+#endif /* NO_WOLFSSH_SERVER */
+
     unitResult = test_BuildNameList_emptySrc();
     printf("BuildNameList_emptySrc: %s\n",
            (unitResult == 0 ? "SUCCESS" : "FAILED"));

wolfssh/ssh.h

@@ -280,6 +280,9 @@ WOLFSSH_API WOLFSSH_CHANNEL* wolfSSH_ChannelFind(WOLFSSH* ssh, word32 id,
         byte peer);
 WOLFSSH_API WOLFSSH_CHANNEL* wolfSSH_ChannelNext(WOLFSSH* ssh,
         WOLFSSH_CHANNEL* channel);
+/* Drains buffered data from the named channel: returns 0 when empty, never
+ * receives, never reports EOF. Carries wolfSSH_stream_read()'s window-adjust
+ * contract but does not clear ssh->error on entry; check it after the call. */
 WOLFSSH_API int wolfSSH_ChannelRead(WOLFSSH_CHANNEL* channel, byte* buf,
         word32 bufSz);
 WOLFSSH_API int wolfSSH_ChannelSend(WOLFSSH_CHANNEL* channel, const byte* buf,
@@ -568,6 +571,9 @@ WOLFSSH_API int wolfSSH_shutdown(WOLFSSH* ssh);
  * dry. A CHANNEL_EOF already received outranks that drain: both report
  * WS_EOF with data possibly still buffered. RFC 4253 section 11.1. */
 WOLFSSH_API int wolfSSH_stream_peek(WOLFSSH* ssh, byte* buf, word32 bufSz);
+/* Returns the bytes read; the next read clears the status. WS_WANT_WRITE
+ * from wolfSSH_get_error() means the adjust is queued; it goes out on the
+ * next send or a wolfSSH_worker() whose receive succeeded. Others failed. */
 WOLFSSH_API int wolfSSH_stream_read(WOLFSSH* ssh, byte* buf, word32 bufSz);
 WOLFSSH_API int wolfSSH_stream_send(WOLFSSH* ssh, byte* buf, word32 bufSz);
 WOLFSSH_API int wolfSSH_stream_exit(WOLFSSH* ssh, int status);
@@ -612,6 +618,9 @@ WOLFSSH_API int wolfSSH_SendIgnore(WOLFSSH* ssh, const byte* buf, word32 bufSz);
 WOLFSSH_API int wolfSSH_SendDisconnect(WOLFSSH* ssh, word32 reason);
 WOLFSSH_API int wolfSSH_global_request(WOLFSSH* ssh, const unsigned char* data,
         word32 dataSz, int reply);
+/* Reads the channel named by channelId, with wolfSSH_ChannelRead()'s
+ * contract, except that wolfSSH_ChannelIdRead() reads during a rekey where
+ * wolfSSH_ChannelRead() returns WS_REKEYING. */
 WOLFSSH_API int wolfSSH_ChannelIdRead(WOLFSSH* ssh, word32 channelId,
         byte* buf, word32 bufSz);
 WOLFSSH_API int wolfSSH_ChannelIdSend(WOLFSSH* ssh, word32 channelId,