From 012499846a4b81cb8c876f5898dbae1004f80e69 Mon Sep 17 00:00:00 2001 From: Yosuke Shimizu Date: Thu, 20 Aug 2026 16:37:21 +0900 Subject: [PATCH 1/3] examples/echoserver: keep the unsent tail of a partial transfer - app_write_all() hands a whole buffer to a local descriptor across partial transfers and returns WS_FATAL_ERROR otherwise; it waits on writability with a bounded try count and checks the select() result. The pty, agent and forwarding writes in ssh_worker() call it, and SOCKET_EAGAIN and SOCKET_EINTR join the socket error macros. - ssh_worker() carries shellBufferIdx and agentBufferIdx beside the existing fwdBufferIdx. A descriptor is read only while its staging buffer is empty and stays out of the read set otherwise; the forwarding recv() fills the buffer from its base. - Each buffer has a flush block, outside the descriptor-state guards, that subtracts what wolfSSH_ChannelIdSend() took and moves the remainder down. WS_CHANNEL_NOT_CONF, WS_CHAN_RXD, WS_WINDOW_FULL and WS_REKEYING hold the data for a later pass. - WS_WANT_WRITE also holds it and sets wantWrite, which adds sshFd to a write set passed to select(). - Echo mode reads the channel into shellCtx.buffer and shares the shell flush block; process_bytes() runs on that buffer. Issue: F-10544 --- examples/echoserver/echoserver.c | 316 ++++++++++++++++++++++++------- 1 file changed, 245 insertions(+), 71 deletions(-) diff --git a/examples/echoserver/echoserver.c b/examples/echoserver/echoserver.c index 2dd03dc8c..559f271f6 100644 --- a/examples/echoserver/echoserver.c +++ b/examples/echoserver/echoserver.c @@ -109,12 +109,16 @@ #define SOCKET_ECONNRESET ECONNRESET #define SOCKET_ECONNABORTED ECONNABORTED #define SOCKET_EWOULDBLOCK EWOULDBLOCK + #define SOCKET_EAGAIN EAGAIN + #define SOCKET_EINTR EINTR #else #include #define SOCKET_ERRNO WSAGetLastError() #define SOCKET_ECONNRESET WSAECONNRESET #define SOCKET_ECONNABORTED WSAECONNABORTED #define SOCKET_EWOULDBLOCK WSAEWOULDBLOCK + #define SOCKET_EAGAIN WSAEWOULDBLOCK + #define SOCKET_EINTR WSAEINTR #endif @@ -800,6 +804,72 @@ static void ChildSig(int sig) } #endif +#if defined(WOLFSSH_SHELL) || defined(WOLFSSH_AGENT) || defined(WOLFSSH_FWD) + +#ifndef EXAMPLE_WRITE_TRIES + #define EXAMPLE_WRITE_TRIES 20 +#endif +#ifndef EXAMPLE_WRITE_WAIT_US + #define EXAMPLE_WRITE_WAIT_US 10000 +#endif + +/* Hand the whole buffer to a local descriptor, advancing past each partial + * transfer. Returns sz, or WS_FATAL_ERROR when it could not be drained. The + * wait only applies if one of these descriptors is made non-blocking. */ +static int app_write_all(WS_SOCKET_T fd, const byte* buf, int sz, int isSocket) +{ + fd_set writeFds; + struct timeval to; + int idx = 0; + int tries = 0; + int cnt; + int err; + int rc; + + if (buf == NULL || sz < 0) { + return WS_FATAL_ERROR; + } + + while (idx < sz && tries < EXAMPLE_WRITE_TRIES) { + cnt = -1; + err = 0; + #ifdef WOLFSSH_SHELL + if (!isSocket) { + cnt = (int)write(fd, buf + idx, (size_t)(sz - idx)); + err = errno; + } + #endif + if (isSocket) { + cnt = (int)send(fd, (const char*)buf + idx, sz - idx, 0); + err = SOCKET_ERRNO; + } + + if (cnt > 0) { + idx += cnt; + tries = 0; + } + else if (cnt < 0 && (err == SOCKET_EINTR || err == SOCKET_EAGAIN || + err == SOCKET_EWOULDBLOCK)) { + FD_ZERO(&writeFds); + FD_SET(fd, &writeFds); + to.tv_sec = 0; + to.tv_usec = EXAMPLE_WRITE_WAIT_US; + rc = select((int)fd + 1, NULL, &writeFds, NULL, &to); + if (rc < 0 && SOCKET_ERRNO != SOCKET_EINTR) { + return WS_FATAL_ERROR; + } + tries++; + } + else { + return WS_FATAL_ERROR; + } + } + + return (idx == sz) ? sz : WS_FATAL_ERROR; +} + +#endif /* WOLFSSH_SHELL || WOLFSSH_AGENT || WOLFSSH_FWD */ + static int ssh_worker(thread_ctx_t* threadCtx) { WOLFSSH* ssh; @@ -879,15 +949,19 @@ static int ssh_worker(thread_ctx_t* threadCtx) #ifdef WOLFSSH_SHELL struct termios tios; #endif + word32 shellBufferIdx = 0; + int shellRxPending = 0; #ifdef WOLFSSH_AGENT WS_SOCKET_T agentFd = -1; WS_SOCKET_T agentListenFd = threadCtx->agentCtx.listenFd; word32 agentChannelId = -1; + word32 agentBufferIdx = 0; #endif #ifdef WOLFSSH_FWD WS_SOCKET_T fwdFd = -1; word32 fwdBufferIdx = 0; #endif + int wantWrite = 0; #ifdef WOLFSSH_SHELL if (!threadCtx->echo) { @@ -938,6 +1012,7 @@ static int ssh_worker(thread_ctx_t* threadCtx) while (ChildRunning) { fd_set readFds; + fd_set writeFds; WS_SOCKET_T maxFd; int cnt_r; int cnt_w; @@ -946,8 +1021,13 @@ static int ssh_worker(thread_ctx_t* threadCtx) FD_SET(sshFd, &readFds); maxFd = sshFd; + FD_ZERO(&writeFds); + if (wantWrite) { + FD_SET(sshFd, &writeFds); + } + #ifdef WOLFSSH_SHELL - if (!threadCtx->echo) { + if (!threadCtx->echo && shellBufferIdx == 0) { FD_SET(childFd, &readFds); if (childFd > maxFd) maxFd = childFd; @@ -959,7 +1039,7 @@ static int ssh_worker(thread_ctx_t* threadCtx) if (agentListenFd > maxFd) maxFd = agentListenFd; } - if (agentFd >= 0 + if (agentFd >= 0 && agentBufferIdx == 0 && threadCtx->agentCtx.state == APP_STATE_CONNECTED) { FD_SET(agentFd, &readFds); if (agentFd > maxFd) @@ -975,17 +1055,18 @@ static int ssh_worker(thread_ctx_t* threadCtx) if (threadCtx->fwdCtx.listenFd > maxFd) maxFd = threadCtx->fwdCtx.listenFd; } - if (fwdFd >= 0 + if (fwdFd >= 0 && fwdBufferIdx == 0 && threadCtx->fwdCtx.state == APP_STATE_CONNECTED) { FD_SET(fwdFd, &readFds); if (fwdFd > maxFd) maxFd = fwdFd; } #endif /* WOLFSSH_FWD */ - rc = select((int)maxFd + 1, &readFds, NULL, NULL, NULL); + rc = select((int)maxFd + 1, &readFds, &writeFds, NULL, NULL); if (rc == -1) { break; } + wantWrite = 0; if (FD_ISSET(sshFd, &readFds)) { word32 lastChannel = 0; @@ -1001,43 +1082,51 @@ static int ssh_worker(thread_ctx_t* threadCtx) rc = wolfSSH_get_error(ssh); if (rc == WS_CHAN_RXD) { if (lastChannel == threadCtx->shellCtx.channelId) { - cnt_r = wolfSSH_ChannelIdRead(ssh, - threadCtx->shellCtx.channelId, - threadCtx->channelBuffer, - sizeof threadCtx->channelBuffer); - if (cnt_r <= 0) - break; - #ifdef SHELL_DEBUG - buf_dump(threadCtx->channelBuffer, cnt_r); - #endif #ifdef WOLFSSH_SHELL - if (!threadCtx->echo) { - cnt_w = (int)write(childFd, - threadCtx->channelBuffer, cnt_r); - } - else { - cnt_w = wolfSSH_ChannelIdSend(ssh, - threadCtx->shellCtx.channelId, - threadCtx->channelBuffer, cnt_r); - if (cnt_r > 0) { - int doStop = process_bytes(threadCtx, - threadCtx->channelBuffer, - cnt_r); - ChildRunning = !doStop; - } - } - #else - cnt_w = wolfSSH_ChannelIdSend(ssh, - threadCtx->shellCtx.channelId, - threadCtx->channelBuffer, cnt_r); - if (cnt_r > 0) { - int doStop = process_bytes(threadCtx, - threadCtx->channelBuffer, cnt_r); - ChildRunning = !doStop; + if (!threadCtx->echo) { + cnt_r = wolfSSH_ChannelIdRead(ssh, + threadCtx->shellCtx.channelId, + threadCtx->channelBuffer, + sizeof threadCtx->channelBuffer); + if (cnt_r <= 0) + break; + #ifdef SHELL_DEBUG + buf_dump(threadCtx->channelBuffer, cnt_r); + #endif + cnt_w = app_write_all(childFd, + threadCtx->channelBuffer, cnt_r, 0); + if (cnt_w <= 0) + break; } + else #endif - if (cnt_w <= 0) - break; + if (shellBufferIdx + < sizeof threadCtx->shellCtx.buffer) { + int doStop; + + shellRxPending = 1; + cnt_r = wolfSSH_ChannelIdRead(ssh, + threadCtx->shellCtx.channelId, + threadCtx->shellCtx.buffer + + shellBufferIdx, + (word32)(sizeof + threadCtx->shellCtx.buffer + - shellBufferIdx)); + if (cnt_r <= 0) + break; + #ifdef SHELL_DEBUG + buf_dump(threadCtx->shellCtx.buffer + + shellBufferIdx, cnt_r); + #endif + doStop = process_bytes(threadCtx, + threadCtx->shellCtx.buffer + + shellBufferIdx, cnt_r); + shellBufferIdx += (word32)cnt_r; + ChildRunning = !doStop; + } + else { + shellRxPending = 1; + } } #ifdef WOLFSSH_AGENT if (lastChannel == agentChannelId) { @@ -1049,8 +1138,8 @@ static int ssh_worker(thread_ctx_t* threadCtx) #ifdef SHELL_DEBUG buf_dump(threadCtx->channelBuffer, cnt_r); #endif - cnt_w = (int)send(agentFd, - threadCtx->channelBuffer, cnt_r, 0); + cnt_w = app_write_all(agentFd, + threadCtx->channelBuffer, cnt_r, 1); if (cnt_w <= 0) break; } @@ -1068,8 +1157,8 @@ static int ssh_worker(thread_ctx_t* threadCtx) #ifdef SHELL_DEBUG buf_dump(threadCtx->channelBuffer, cnt_r); #endif - cnt_w = (int)send(fwdFd, threadCtx->channelBuffer, - cnt_r, 0); + cnt_w = app_write_all(fwdFd, + threadCtx->channelBuffer, cnt_r, 1); if (cnt_w <= 0) break; } @@ -1090,6 +1179,9 @@ static int ssh_worker(thread_ctx_t* threadCtx) NULL, 0); threadCtx->fwdCbCtx.originName = NULL; } + /* The channel is gone, so drop anything still + * staged for it */ + fwdBufferIdx = 0; threadCtx->fwdCtx.state = APP_STATE_LISTEN; } #endif @@ -1106,7 +1198,7 @@ static int ssh_worker(thread_ctx_t* threadCtx) } #ifdef WOLFSSH_SHELL - if (!threadCtx->echo) { + if (!threadCtx->echo && shellBufferIdx == 0) { if (FD_ISSET(childFd, &readFds)) { cnt_r = (int)read(childFd, threadCtx->shellCtx.buffer, @@ -1127,19 +1219,64 @@ static int ssh_worker(thread_ctx_t* threadCtx) #ifdef SHELL_DEBUG buf_dump(threadCtx->shellCtx.buffer, cnt_r); #endif - if (cnt_r > 0) { - cnt_w = wolfSSH_ChannelIdSend(ssh, - threadCtx->shellCtx.channelId, - threadCtx->shellCtx.buffer, cnt_r); - if (cnt_w < 0) - break; - } + shellBufferIdx = (word32)cnt_r; } } } #endif /* WOLFSSH_SHELL */ + /* Drain what a full staging buffer made us defer. A zero read + * means the channel is empty, not an error. */ + if (shellRxPending + && shellBufferIdx < sizeof threadCtx->shellCtx.buffer) { + cnt_r = wolfSSH_ChannelIdRead(ssh, + threadCtx->shellCtx.channelId, + threadCtx->shellCtx.buffer + shellBufferIdx, + (word32)(sizeof threadCtx->shellCtx.buffer + - shellBufferIdx)); + if (cnt_r < 0) + break; + if (cnt_r == 0) { + shellRxPending = 0; + } + else { + ChildRunning = !process_bytes(threadCtx, + threadCtx->shellCtx.buffer + shellBufferIdx, + cnt_r); + shellBufferIdx += (word32)cnt_r; + } + } + if (shellBufferIdx > 0) { + /* A send clamped by a max packet size returns short but + * positive, and the rest can go out now */ + do { + cnt_w = wolfSSH_ChannelIdSend(ssh, + threadCtx->shellCtx.channelId, + threadCtx->shellCtx.buffer, shellBufferIdx); + if (cnt_w > 0) { + shellBufferIdx -= (word32)cnt_w; + if (shellBufferIdx > 0) { + WMEMMOVE(threadCtx->shellCtx.buffer, + threadCtx->shellCtx.buffer + cnt_w, + shellBufferIdx); + } + } + } while (cnt_w > 0 && shellBufferIdx > 0); + + if (cnt_w > 0) { + /* drained */ + } + else if (cnt_w == WS_WANT_WRITE) { + wantWrite = 1; + } + else if (cnt_w != WS_CHANNEL_NOT_CONF + && cnt_w != WS_CHAN_RXD + && cnt_w != WS_WINDOW_FULL + && cnt_w != WS_REKEYING) { + break; + } + } #ifdef WOLFSSH_AGENT - if (agentFd >= 0 + if (agentFd >= 0 && agentBufferIdx == 0 && threadCtx->agentCtx.state == APP_STATE_CONNECTED) { if (FD_ISSET(agentFd, &readFds)) { #ifdef SHELL_DEBUG @@ -1173,12 +1310,35 @@ static int ssh_worker(thread_ctx_t* threadCtx) #ifdef SHELL_DEBUG buf_dump(threadCtx->agentCtx.buffer, cnt_r); #endif - cnt_w = wolfSSH_ChannelIdSend(ssh, agentChannelId, - threadCtx->agentCtx.buffer, cnt_r); - if (cnt_w <= 0) { - break; + agentBufferIdx = (word32)cnt_r; + } + } + } + if (agentBufferIdx > 0) { + do { + cnt_w = wolfSSH_ChannelIdSend(ssh, agentChannelId, + threadCtx->agentCtx.buffer, agentBufferIdx); + if (cnt_w > 0) { + agentBufferIdx -= (word32)cnt_w; + if (agentBufferIdx > 0) { + WMEMMOVE(threadCtx->agentCtx.buffer, + threadCtx->agentCtx.buffer + cnt_w, + agentBufferIdx); } } + } while (cnt_w > 0 && agentBufferIdx > 0); + + if (cnt_w > 0) { + /* drained */ + } + else if (cnt_w == WS_WANT_WRITE) { + wantWrite = 1; + } + else if (cnt_w != WS_CHANNEL_NOT_CONF + && cnt_w != WS_CHAN_RXD + && cnt_w != WS_WINDOW_FULL + && cnt_w != WS_REKEYING) { + break; } } if (threadCtx->agentCtx.state == APP_STATE_LISTEN) { @@ -1201,15 +1361,14 @@ static int ssh_worker(thread_ctx_t* threadCtx) } #endif /* WOLFSSH_AGENT */ #ifdef WOLFSSH_FWD - if (fwdFd >= 0 + if (fwdFd >= 0 && fwdBufferIdx == 0 && threadCtx->fwdCtx.state == APP_STATE_CONNECTED) { if (FD_ISSET(fwdFd, &readFds)) { #ifdef SHELL_DEBUG printf("fwdFd set in readfd\n"); #endif - cnt_r = (int)recv(fwdFd, - threadCtx->fwdCtx.buffer + fwdBufferIdx, - sizeof threadCtx->fwdCtx.buffer - fwdBufferIdx, 0); + cnt_r = (int)recv(fwdFd, threadCtx->fwdCtx.buffer, + sizeof threadCtx->fwdCtx.buffer, 0); if (cnt_r == 0) { /* Read zero-returned. Socket is closed. Go back to listening. */ @@ -1244,25 +1403,40 @@ static int ssh_worker(thread_ctx_t* threadCtx) #ifdef SHELL_DEBUG buf_dump(threadCtx->fwdCtx.buffer, cnt_r); #endif - fwdBufferIdx += cnt_r; + fwdBufferIdx = (word32)cnt_r; } } - if (fwdBufferIdx > 0) { + } + if (fwdBufferIdx > 0 + && threadCtx->fwdCtx.state == APP_STATE_CONNECTED) { + do { cnt_w = wolfSSH_ChannelIdSend(ssh, threadCtx->fwdCtx.channelId, threadCtx->fwdCtx.buffer, fwdBufferIdx); if (cnt_w > 0) { - fwdBufferIdx = 0; - } - else if (cnt_w == WS_CHANNEL_NOT_CONF || - cnt_w == WS_CHAN_RXD) { - #ifdef SHELL_DEBUG - printf("Waiting for channel open confirmation.\n"); - #endif - } - else { - break; + fwdBufferIdx -= (word32)cnt_w; + if (fwdBufferIdx > 0) { + WMEMMOVE(threadCtx->fwdCtx.buffer, + threadCtx->fwdCtx.buffer + cnt_w, + fwdBufferIdx); + } } + } while (cnt_w > 0 && fwdBufferIdx > 0); + + if (cnt_w > 0) { + /* drained */ + } + else if (cnt_w == WS_CHANNEL_NOT_CONF + || cnt_w == WS_CHAN_RXD) { + #ifdef SHELL_DEBUG + printf("Waiting for channel open confirmation.\n"); + #endif + } + else if (cnt_w == WS_WANT_WRITE) { + wantWrite = 1; + } + else if (cnt_w != WS_WINDOW_FULL && cnt_w != WS_REKEYING) { + break; } } if (threadCtx->fwdCtx.state == APP_STATE_LISTEN From 7d8a13d5f161bc0a1c195cc4eeecfffb3c75866a Mon Sep 17 00:00:00 2001 From: Yosuke Shimizu Date: Thu, 20 Aug 2026 16:37:35 +0900 Subject: [PATCH 2/3] ssh.c: credit the channel window for the bytes a read consumes - wolfSSH_stream_read() advances inputBuffer->idx before _UpdateChannelWindow(), records a non-success result in ssh->error and reports the byte count. - tests/api.c adds test_wolfSSH_stream_read_WindowCredit(), which round-trips 2000 bytes through a 1024-byte client receive window, with its own user-auth and host-key callbacks. - tests/unit.c adds test_stream_read_deferredWindowAdjust(), which puts a full window of channel data and reads it back with an IO send that reports WS_CBIO_ERR_WANT_WRITE, checking the byte count, the payload, ssh->error, the credited window and the consumed buffer. --- src/ssh.c | 7 +- tests/api.c | 177 +++++++++++++++++++++++++++++++++++++++++++++++++++ tests/unit.c | 68 ++++++++++++++++++++ 3 files changed, 249 insertions(+), 3 deletions(-) diff --git a/src/ssh.c b/src/ssh.c index 1f3f7cc49..4dc97a300 100644 --- a/src/ssh.c +++ b/src/ssh.c @@ -1288,11 +1288,12 @@ 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; } + ret = n; } } diff --git a/tests/api.c b/tests/api.c index e432fdc2a..e7379d7fc 100644 --- a/tests/api.c +++ b/tests/api.c @@ -7433,6 +7433,180 @@ static void test_wolfSSH_KeyboardInteractive(void) { ; } #endif /* WOLFSSH_SFTP && !NO_WOLFSSH_CLIENT && !SINGLE_THREADED */ #endif /* WOLFSSH_KEYBOARD_INTERACTIVE */ + +#if defined(WOLFSSH_TEST_ECHOSERVER) && !defined(NO_WOLFSSH_CLIENT) && \ + !defined(SINGLE_THREADED) + +/* Payload outruns both the window and the echo server's EXAMPLE_BUFFER_SZ, so + * the transfer needs mid-stream credit and staging over many passes. Capping + * the max packet size instead would bound the sender's own writes too. */ +#define WINCREDIT_PAYLOAD_SZ 16000 +#define WINCREDIT_WINDOW_SZ 1024 +/* Consecutive stalls, not total passes. */ +#define WINCREDIT_MAX_TRIES 30 + +static byte winCreditPassword[32]; + +static int winCreditUserAuth(byte authType, WS_UserAuthData* authData, + void* ctx) +{ + const char* password = (const char*)ctx; + word32 passwordSz; + + if (authType != WOLFSSH_USERAUTH_PASSWORD || password == NULL) { + return WOLFSSH_USERAUTH_INVALID_AUTHTYPE; + } + + passwordSz = (word32)WSTRLEN(password); + if (passwordSz > (word32)sizeof(winCreditPassword)) { + return WOLFSSH_USERAUTH_INVALID_AUTHTYPE; + } + WMEMCPY(winCreditPassword, password, passwordSz); + authData->sf.password.password = winCreditPassword; + authData->sf.password.passwordSz = passwordSz; + + return WOLFSSH_USERAUTH_SUCCESS; +} + +static int winCreditAcceptKey(const byte* pubKey, word32 pubKeySz, void* ctx) +{ + (void)pubKey; + (void)pubKeySz; + (void)ctx; + return 0; +} + +/* A receive window smaller than the payload must still round-trip: reading + * has to credit the window back so the rest can follow. */ +static void test_wolfSSH_stream_read_WindowCredit(void) +{ + func_args ser; + tcp_ready ready; + THREAD_TYPE serThread; + WOLFSSH_CTX* ctx = NULL; + WOLFSSH* ssh = NULL; + WS_SOCKET_T sockFd = WOLFSSH_SOCKET_INVALID; + SOCKADDR_IN_T clientAddr; + socklen_t clientAddrSz = sizeof(clientAddr); + const char* args[6]; + const char* password = "upthehill"; + byte* payload = NULL; + byte* rxBuf = NULL; + int argsCount; + int i; + int ret; + int err; + int idx; + int tries; + + /* Off the stack: Zephyr runs this on its 32 KB main thread. */ + payload = (byte*)WMALLOC(WINCREDIT_PAYLOAD_SZ, NULL, DYNTYPE_BUFFER); + rxBuf = (byte*)WMALLOC(WINCREDIT_PAYLOAD_SZ, NULL, DYNTYPE_BUFFER); + AssertNotNull(payload); + AssertNotNull(rxBuf); + + WMEMSET(&ser, 0, sizeof(func_args)); + argsCount = 0; + args[argsCount++] = "echoserver"; + args[argsCount++] = "-1"; + args[argsCount++] = "-f"; + args[argsCount++] = "-p"; + args[argsCount++] = "0"; + ser.argv = (char**)args; + ser.argc = argsCount; + ser.signal = &ready; + InitTcpReady(ser.signal); + ThreadStart(echoserver_test, (void*)&ser, &serThread); + WaitTcpReady(&ready); + + ctx = wolfSSH_CTX_new(WOLFSSH_ENDPOINT_CLIENT, NULL); + AssertNotNull(ctx); + AssertIntEQ(wolfSSH_CTX_SetWindowPacketSize(ctx, WINCREDIT_WINDOW_SZ, 0), + WS_SUCCESS); + wolfSSH_CTX_SetPublicKeyCheck(ctx, winCreditAcceptKey); + wolfSSH_SetUserAuth(ctx, winCreditUserAuth); + + ssh = wolfSSH_new(ctx); + AssertNotNull(ssh); + wolfSSH_SetUserAuthCtx(ssh, (void*)password); + AssertIntEQ(wolfSSH_SetUsername(ssh, "jill"), WS_SUCCESS); + + build_addr(&clientAddr, (char*)wolfSshIp, ready.port); + tcp_socket(&sockFd, ((struct sockaddr_in*)&clientAddr)->sin_family); + AssertIntEQ(connect(sockFd, (const struct sockaddr*)&clientAddr, + clientAddrSz), 0); + AssertIntEQ(wolfSSH_set_fd(ssh, (int)sockFd), WS_SUCCESS); + AssertIntEQ(wolfSSH_connect(ssh), WS_SUCCESS); + + /* Lower case only: 0x03, 0x05 and 0x06 are echo server triggers. */ + for (i = 0; i < WINCREDIT_PAYLOAD_SZ; i++) { + payload[i] = (byte)('a' + (i % 26)); + } + + idx = 0; + for (tries = 0; tries < WINCREDIT_MAX_TRIES && idx < WINCREDIT_PAYLOAD_SZ; + tries++) { + ret = wolfSSH_stream_send(ssh, payload + idx, + (word32)(WINCREDIT_PAYLOAD_SZ - idx)); + if (ret > 0) { + idx += ret; + tries = 0; + } + else { + err = wolfSSH_get_error(ssh); + if (err != WS_WANT_READ && err != WS_WANT_WRITE && + err != WS_WINDOW_FULL && err != WS_REKEYING) { + break; + } + tcp_select(sockFd, 1); + } + } + AssertIntEQ(idx, WINCREDIT_PAYLOAD_SZ); + + /* Send blocking: a positive stream_send() only stages the bytes, so a + * short socket write would park the tail. Read non-blocking to bound it. */ + tcp_set_nonblocking(&sockFd); + + /* Fails instead of hanging if a regression drops the echoed tail. */ + idx = 0; + for (tries = 0; tries < WINCREDIT_MAX_TRIES && idx < WINCREDIT_PAYLOAD_SZ; + tries++) { + ret = wolfSSH_stream_read(ssh, rxBuf + idx, + (word32)(WINCREDIT_PAYLOAD_SZ - idx)); + if (ret > 0) { + idx += ret; + tries = 0; + } + else { + err = wolfSSH_get_error(ssh); + if (err != WS_WANT_READ && err != WS_WANT_WRITE && + err != WS_CHAN_RXD && err != WS_REKEYING) { + break; + } + tcp_select(sockFd, 1); + } + } + AssertIntEQ(idx, WINCREDIT_PAYLOAD_SZ); + AssertIntEQ(WMEMCMP(rxBuf, payload, WINCREDIT_PAYLOAD_SZ), 0); + + wolfSSH_shutdown(ssh); + WCLOSESOCKET(sockFd); + wolfSSH_free(ssh); + wolfSSH_CTX_free(ctx); +#ifdef WOLFSSH_ZEPHYR + /* Weird deadlock without this sleep */ + k_sleep(Z_TIMEOUT_TICKS(100)); +#endif + ThreadJoin(serThread); + FreeTcpReady(&ready); + WFREE(payload, NULL, DYNTYPE_BUFFER); + WFREE(rxBuf, NULL, DYNTYPE_BUFFER); +} + +#else /* WOLFSSH_TEST_ECHOSERVER && !NO_WOLFSSH_CLIENT && !SINGLE_THREADED */ +static void test_wolfSSH_stream_read_WindowCredit(void) { ; } +#endif /* WOLFSSH_TEST_ECHOSERVER && !NO_WOLFSSH_CLIENT && !SINGLE_THREADED */ + #endif /* WOLFSSH_TEST_BLOCK */ @@ -7556,6 +7730,9 @@ int wolfSSH_ApiTest(int argc, char** argv) test_wolfSSH_SFTP_SetDefaultPath(); test_wolfSSH_SFTP_SaveOfst(); + /* Channel data flow; needs the echoserver, so it is gated the same way */ + test_wolfSSH_stream_read_WindowCredit(); + /* Either SCP or SFTP */ test_wolfSSH_RealPath(); AssertIntEQ(wolfSSH_Cleanup(), WS_SUCCESS); diff --git a/tests/unit.c b/tests/unit.c index e1eff261c..527f6ec58 100644 --- a/tests/unit.c +++ b/tests/unit.c @@ -6063,6 +6063,69 @@ static int test_SendChannelData_zeroPeerMaxPacket(void) return result; } +/* wolfSSH_stream_read() counterpart of test_ChannelExtDataCreditWantWrite(): + * a deferred credit must not cost the caller the bytes already consumed. */ +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; + } + +done: + wolfSSH_free(ssh); + wolfSSH_CTX_free(ctx); + return result; +} + /* 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. */ @@ -16400,6 +16463,11 @@ int wolfSSH_UnitTest(int argc, char** argv) (unitResult == 0 ? "SUCCESS" : "FAILED")); testResult = testResult || unitResult; + unitResult = test_stream_read_deferredWindowAdjust(); + printf("stream_read_deferredWindowAdjust: %s\n", + (unitResult == 0 ? "SUCCESS" : "FAILED")); + testResult = testResult || unitResult; + unitResult = test_BuildNameList_emptySrc(); printf("BuildNameList_emptySrc: %s\n", (unitResult == 0 ? "SUCCESS" : "FAILED")); From 45b5130ab2417e976fe18b8c64de7be794530b17 Mon Sep 17 00:00:00 2001 From: Yosuke Shimizu Date: Mon, 24 Aug 2026 09:28:53 +0900 Subject: [PATCH 3/3] ssh.c: report the byte count from a channel read that defers its credit - _ChannelRead() records a non-WS_SUCCESS _UpdateChannelWindow() result in channel->ssh->error and returns the bytes copied, logging anything other than WS_WANT_WRITE. - tests/unit.c gains test_ChannelIdRead_deferredWindowAdjust(), which seeds ssh->error, then reads a full window through wolfSSH_ChannelIdRead() with an IO send that reports WS_CBIO_ERR_WANT_WRITE, and checks the byte count, the payload, the recorded error, the local window credit and the drained input buffer. --- src/ssh.c | 16 +++++++++-- tests/unit.c | 79 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+), 3 deletions(-) diff --git a/src/ssh.c b/src/ssh.c index 4dc97a300..6d268ce1e 100644 --- a/src/ssh.c +++ b/src/ssh.c @@ -3801,6 +3801,8 @@ static int _UpdateChannelWindow(WOLFSSH_CHANNEL* channel) } +/* Drains buffered channel data and credits the window for the bytes taken. + * Always reports the bytes copied */ static int _ChannelRead(WOLFSSH_CHANNEL* channel, byte* buf, word32 bufSz) { WOLFSSH_BUFFER* inputBuffer; @@ -3815,10 +3817,18 @@ static int _ChannelRead(WOLFSSH_CHANNEL* channel, byte* buf, word32 bufSz) inputBuffer->idx += bufSz; updateResult = _UpdateChannelWindow(channel); - if (updateResult == WS_SUCCESS) - updateResult = bufSz; + if (updateResult != WS_SUCCESS) { + /* SendPacket() sets ssh->error only for WS_WANT_WRITE, so hard + * failures must be recorded here or they stay hidden. */ + channel->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; } diff --git a/tests/unit.c b/tests/unit.c index 527f6ec58..8e62060ca 100644 --- a/tests/unit.c +++ b/tests/unit.c @@ -6063,6 +6063,8 @@ 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. */ static int test_stream_read_deferredWindowAdjust(void) @@ -6126,6 +6128,76 @@ static int test_stream_read_deferredWindowAdjust(void) return result; } +/* wolfSSH_ChannelIdRead() counterpart of + * test_stream_read_deferredWindowAdjust(): the echoserver's shell, agent and + * forwarding paths all break out on a non-positive read. */ +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; + } + +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. */ @@ -16463,11 +16535,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"));