From 5df52456e2ab95bbc0df320acc47457c52430aae Mon Sep 17 00:00:00 2001 From: Yosuke Shimizu Date: Fri, 18 Sep 2026 11:39:09 +0900 Subject: [PATCH 1/3] examples/echoserver: hold what a channel send did not take - WS_AppCtx gains bufferIdx and bufferOff; app_staged() reports the bytes its buffer still owes the channel. - app_drain_to_channel() advances bufferOff by what wolfSSH_ChannelIdSend() took, keeps the rest on WS_WANT_WRITE, WS_WINDOW_FULL, WS_REKEYING, WS_CHANNEL_NOT_CONF and WS_CHAN_RXD, and raises wantWrite for the first of those. For the other four it puts back the error code it found on entry, so the status of a send that is only owed does not outlive the call. A send reporting more than it was given, or zero, ends the session. - The shell, agent and forward reads stage into their own WS_AppCtx buffer and drain through it; their descriptors leave the select read set while bufferOff is short of bufferIdx. - app_echo_pump() reads the shell channel into shellCtx.buffer, hands each chunk to process_bytes() and drains it back. It reports the channel dry on a zero read alone, and a negative one stops the pump without ending the session. ssh_worker() runs it on every pass rather than on a data report. - process_bytes() sees every echoed chunk, including the backlog that goes out behind the peer's EOF, and only ever clears ChildRunning. - The EOF reply takes its drained state from app_echo_pump(); eofBuffer, eofOff and eofRead are dropped. - app_write_all() writes a whole buffer to the pty, agent and forward descriptors, retrying an interrupted write, in place of the single write() and send() calls on the channel-to-descriptor paths. - The agent and forward contexts clear their staging as they enter a connection; the forward's recv() fills the buffer from the start, and fwdBufferIdx and cnt_w are dropped. Issue: F-10544 --- examples/echoserver/echoserver.c | 320 ++++++++++++++++++++----------- 1 file changed, 210 insertions(+), 110 deletions(-) diff --git a/examples/echoserver/echoserver.c b/examples/echoserver/echoserver.c index 4dde815b6..131677110 100644 --- a/examples/echoserver/echoserver.c +++ b/examples/echoserver/echoserver.c @@ -185,6 +185,10 @@ typedef struct WS_AppCtx { word32 channelId; WS_AppState state; byte buffer[EXAMPLE_BUFFER_SZ]; + /* Bytes staged in buffer and how many of them the channel has taken, + * with 0 <= bufferOff <= bufferIdx. Both survive a worker pass. */ + word32 bufferIdx; + word32 bufferOff; } WS_AppCtx; @@ -236,9 +240,6 @@ typedef struct { int doScp; #endif byte channelBuffer[EXAMPLE_BUFFER_SZ]; - /* The EOF drain holds an unsent tail across worker passes, - * so it cannot share channelBuffer with the read path. */ - byte eofBuffer[EXAMPLE_BUFFER_SZ]; char statsBuffer[EXAMPLE_BUFFER_SZ]; } thread_ctx_t; @@ -1113,6 +1114,146 @@ static void buf_dump(unsigned char *buf, int len) #endif +/* Bytes appCtx still owes the channel. */ +static word32 app_staged(const WS_AppCtx* appCtx) +{ + return appCtx->bufferIdx - appCtx->bufferOff; +} + + +/* Hand the staged bytes to the channel, advancing bufferOff by however many + * it took. Returns 0 while the send is owed or done, negative to end the + * session. */ +static int app_drain_to_channel(WOLFSSH* ssh, WS_AppCtx* appCtx, + word32 channelId, int* wantWrite) +{ + int savedError = ssh->error; + int ret = 0; + int cnt; + + while (app_staged(appCtx) > 0) { + cnt = wolfSSH_ChannelIdSend(ssh, channelId, + appCtx->buffer + appCtx->bufferOff, app_staged(appCtx)); + if (cnt > 0) { + if ((word32)cnt > app_staged(appCtx)) { + ret = WS_FATAL_ERROR; + break; + } + appCtx->bufferOff += (word32)cnt; + continue; + } + + if (cnt == WS_WANT_WRITE) { + *wantWrite = 1; + } + else if (cnt == WS_WINDOW_FULL || cnt == WS_REKEYING + || cnt == WS_CHANNEL_NOT_CONF || cnt == WS_CHAN_RXD) { + /* The send is owed, not failed, so put back the code */ + ssh->error = savedError; + } + else { + /* Zero is not allowed here */ + ret = (cnt < 0) ? cnt : WS_FATAL_ERROR; + } + break; + } + + if (app_staged(appCtx) == 0) { + appCtx->bufferIdx = 0; + appCtx->bufferOff = 0; + } + + return ret; +} + + +/* Loop the shell channel's buffered data back to it, staging in + * shellCtx.buffer, which echo mode leaves free. Sets dry once the channel + * holds nothing more. Returns 0, or negative to end the session. */ +static int app_echo_pump(WOLFSSH* ssh, thread_ctx_t* threadCtx, int* wantWrite, + int* dry) +{ + WS_AppCtx* appCtx = &threadCtx->shellCtx; + int cnt; + int ret; + + *dry = 0; + + for (;;) { + if (app_staged(appCtx) == 0) { + cnt = wolfSSH_ChannelIdRead(ssh, appCtx->channelId, + appCtx->buffer, (word32)sizeof appCtx->buffer); + /* Only a zero read is dry; a negative one just stops the pump. */ + if (cnt <= 0) { + *dry = (cnt == 0); + break; + } + #ifdef SHELL_DEBUG + buf_dump(appCtx->buffer, cnt); + #endif + appCtx->bufferIdx = (word32)cnt; + appCtx->bufferOff = 0; + if (process_bytes(threadCtx, appCtx->buffer, cnt)) { + ChildRunning = 0; + } + } + + ret = app_drain_to_channel(ssh, appCtx, appCtx->channelId, wantWrite); + if (ret < 0) { + return ret; + } + if (app_staged(appCtx) > 0) { + break; + } + } + + return 0; +} + + +#if defined(WOLFSSH_SHELL) || defined(WOLFSSH_AGENT) || defined(WOLFSSH_FWD) + +/* Write every byte to a blocking descriptor, retrying what an interrupted + * write left behind. Returns bufSz, or -1 */ +static int app_write_all(WS_SOCKET_T fd, const byte* buf, word32 bufSz, + int isSocket) +{ + word32 off = 0; + int cnt; + +#ifndef WOLFSSH_SHELL + (void)isSocket; +#endif + + while (off < bufSz) { +#ifdef WOLFSSH_SHELL + if (!isSocket) { + cnt = (int)write(fd, buf + off, bufSz - off); + } + else { + cnt = (int)send(fd, (const char*)buf + off, (int)(bufSz - off), 0); + } +#else + cnt = (int)send(fd, (const char*)buf + off, (int)(bufSz - off), 0); +#endif + + if (cnt > 0) { + off += (word32)cnt; + } + else if (cnt < 0 && SOCKET_ERRNO == SOCKET_EINTR) { + continue; + } + else { + return -1; + } + } + + return (int)bufSz; +} + +#endif /* WOLFSSH_SHELL || WOLFSSH_AGENT || WOLFSSH_FWD */ + + static int ssh_worker(thread_ctx_t* threadCtx) { WOLFSSH* ssh; @@ -1122,9 +1263,6 @@ static int ssh_worker(thread_ctx_t* threadCtx) * still leaves through the cleanup below it. */ int workerRet = 0; int eofAnswered = 0; - /* Held across passes with 0 <= eofOff <= eofRead. */ - int eofRead = 0; - int eofOff = 0; /* Without a shell there is no child to outlive the peer's EOF, and the * read path echoes unconditionally. */ int echoOnly = 1; @@ -1187,7 +1325,6 @@ static int ssh_worker(thread_ctx_t* threadCtx) #endif #ifdef WOLFSSH_FWD WS_SOCKET_T fwdFd = -1; - word32 fwdBufferIdx = 0; #endif ChildRunning = 1; @@ -1198,7 +1335,6 @@ static int ssh_worker(thread_ctx_t* threadCtx) int writable; WS_SOCKET_T maxFd; int cnt_r; - int cnt_w; FD_ZERO(&readFds); FD_SET(sshFd, &readFds); @@ -1229,9 +1365,12 @@ static int ssh_worker(thread_ctx_t* threadCtx) if (wantWrite) FD_SET(sshFd, &writeFds); + /* Keep a descriptor out of the read set while its buffer is not + * empty */ #ifdef WOLFSSH_SHELL if (threadCtx->shellCtx.state == APP_STATE_CONNECTED - && threadCtx->shellCtx.appFd >= 0) { + && threadCtx->shellCtx.appFd >= 0 + && app_staged(&threadCtx->shellCtx) == 0) { FD_SET(threadCtx->shellCtx.appFd, &readFds); if (threadCtx->shellCtx.appFd > maxFd) maxFd = threadCtx->shellCtx.appFd; @@ -1247,7 +1386,8 @@ static int ssh_worker(thread_ctx_t* threadCtx) maxFd = threadCtx->agentCtx.listenFd; } if (agentFd >= 0 - && threadCtx->agentCtx.state == APP_STATE_CONNECTED) { + && threadCtx->agentCtx.state == APP_STATE_CONNECTED + && app_staged(&threadCtx->agentCtx) == 0) { FD_SET(agentFd, &readFds); if (agentFd > maxFd) maxFd = agentFd; @@ -1263,7 +1403,8 @@ static int ssh_worker(thread_ctx_t* threadCtx) maxFd = threadCtx->fwdCtx.listenFd; } if (fwdFd >= 0 - && threadCtx->fwdCtx.state == APP_STATE_CONNECTED) { + && threadCtx->fwdCtx.state == APP_STATE_CONNECTED + && app_staged(&threadCtx->fwdCtx) == 0) { FD_SET(fwdFd, &readFds); if (fwdFd > maxFd) maxFd = fwdFd; @@ -1324,39 +1465,11 @@ static int ssh_worker(thread_ctx_t* threadCtx) threadCtx->shellCtx.channelId, WS_CHANNEL_ID_SELF); if (eofChannel != NULL && wolfSSH_ChannelGetEof(eofChannel)) { - int eofSent; int eofDrained = 0; - for (;;) { - /* A send is bounded by the peer's window and - * packet size, so a short one is normal. Read - * the next chunk only once the last one is out: - * the read consumed it from the channel, so its - * tail cannot be dropped. */ - if (eofOff == eofRead) { - int eofRxd; - - eofOff = eofRead = 0; - eofRxd = wolfSSH_ChannelIdRead(ssh, - threadCtx->shellCtx.channelId, - threadCtx->eofBuffer, - sizeof threadCtx->eofBuffer); - /* A negative read is a rekey or a stalled - * channel, not a drained one. */ - if (eofRxd <= 0) { - eofDrained = (eofRxd == 0); - break; - } - eofRead = eofRxd; - } - - eofSent = wolfSSH_ChannelIdSend(ssh, - threadCtx->shellCtx.channelId, - threadCtx->eofBuffer + eofOff, - eofRead - eofOff); - if (eofSent <= 0) - break; - eofOff += eofSent; + if (app_echo_pump(ssh, threadCtx, &wantWrite, + &eofDrained) < 0) { + break; } /* Only an emptied channel earns the EOF; anything @@ -1384,8 +1497,14 @@ static int ssh_worker(thread_ctx_t* threadCtx) * wolfSSH_ChannelIdRead() has no isKeying gate; the window * credit it owes is parked until the rekey finishes. */ if (rc == WS_CHAN_RXD || rc == WS_REKEYING) { - if (threadCtx->shellCtx.state == APP_STATE_CONNECTED && - lastChannel == threadCtx->shellCtx.channelId) { + #ifdef WOLFSSH_SHELL + /* Echo mode is served by app_echo_pump(), which runs + * on every pass rather than on a data report. */ + if (!echoOnly + && threadCtx->shellCtx.state + == APP_STATE_CONNECTED + && lastChannel + == threadCtx->shellCtx.channelId) { cnt_r = wolfSSH_ChannelIdRead(ssh, threadCtx->shellCtx.channelId, threadCtx->channelBuffer, @@ -1400,36 +1519,13 @@ static int ssh_worker(thread_ctx_t* threadCtx) #ifdef SHELL_DEBUG buf_dump(threadCtx->channelBuffer, cnt_r); #endif - #ifdef WOLFSSH_SHELL - if (!threadCtx->echo) { - cnt_w = (int)write( - threadCtx->shellCtx.appFd, - 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; - } - #endif - if (cnt_w <= 0) + if (app_write_all(threadCtx->shellCtx.appFd, + threadCtx->channelBuffer, + (word32)cnt_r, 0) < 0) { break; + } } + #endif /* WOLFSSH_SHELL */ #ifdef WOLFSSH_AGENT if (lastChannel == agentChannelId) { cnt_r = wolfSSH_ChannelIdRead(ssh, agentChannelId, @@ -1445,9 +1541,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); - if (cnt_w <= 0) + if (app_write_all(agentFd, threadCtx->channelBuffer, + (word32)cnt_r, 1) < 0) break; } #endif @@ -1469,9 +1564,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); - if (cnt_w <= 0) + if (app_write_all(fwdFd, threadCtx->channelBuffer, + (word32)cnt_r, 1) < 0) break; } #endif @@ -1556,17 +1650,29 @@ 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; - } + threadCtx->shellCtx.bufferIdx = (word32)cnt_r; + threadCtx->shellCtx.bufferOff = 0; } } } #endif /* WOLFSSH_SHELL */ + /* Retry a tail the channel did not take, whatever woke this + * pass. */ + if (threadCtx->shellCtx.state == APP_STATE_CONNECTED) { + int echoDry; + + if (!echoOnly) { + if (app_drain_to_channel(ssh, &threadCtx->shellCtx, + threadCtx->shellCtx.channelId, + &wantWrite) < 0) { + break; + } + } + else if (app_echo_pump(ssh, threadCtx, &wantWrite, + &echoDry) < 0) { + break; + } + } #ifdef WOLFSSH_AGENT if (agentFd >= 0 && threadCtx->agentCtx.state == APP_STATE_CONNECTED) { @@ -1608,13 +1714,14 @@ 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; - } + threadCtx->agentCtx.bufferIdx = (word32)cnt_r; + threadCtx->agentCtx.bufferOff = 0; } } + if (app_drain_to_channel(ssh, &threadCtx->agentCtx, + agentChannelId, &wantWrite) < 0) { + break; + } } if (threadCtx->agentCtx.state == APP_STATE_LISTEN && threadCtx->agentCtx.listenFd >= 0) { @@ -1630,6 +1737,8 @@ static int ssh_worker(thread_ctx_t* threadCtx) } } else { + threadCtx->agentCtx.bufferIdx = 0; + threadCtx->agentCtx.bufferOff = 0; threadCtx->agentCtx.state = APP_STATE_CONNECTED; threadCtx->agentCtx.appFd = agentFd; } @@ -1643,9 +1752,8 @@ static int ssh_worker(thread_ctx_t* threadCtx) #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. */ @@ -1682,25 +1790,13 @@ static int ssh_worker(thread_ctx_t* threadCtx) #ifdef SHELL_DEBUG buf_dump(threadCtx->fwdCtx.buffer, cnt_r); #endif - fwdBufferIdx += cnt_r; + threadCtx->fwdCtx.bufferIdx = (word32)cnt_r; + threadCtx->fwdCtx.bufferOff = 0; } } - if (fwdBufferIdx > 0) { - 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; - } + if (app_drain_to_channel(ssh, &threadCtx->fwdCtx, + threadCtx->fwdCtx.channelId, &wantWrite) < 0) { + break; } } if (threadCtx->fwdCtx.state == APP_STATE_LISTEN @@ -1762,6 +1858,8 @@ static int ssh_worker(thread_ctx_t* threadCtx) threadCtx->fwdCbCtx.originName, threadCtx->fwdCbCtx.originPort); if (newChannel != NULL) { + threadCtx->fwdCtx.bufferIdx = 0; + threadCtx->fwdCtx.bufferOff = 0; threadCtx->fwdCtx.state = APP_STATE_CONNECTED; } } @@ -1770,6 +1868,8 @@ static int ssh_worker(thread_ctx_t* threadCtx) threadCtx->fwdCbCtx.hostPort); if (fwdFd > 0) { + threadCtx->fwdCtx.bufferIdx = 0; + threadCtx->fwdCtx.bufferOff = 0; threadCtx->fwdCtx.appFd = fwdFd; threadCtx->fwdCtx.state = APP_STATE_CONNECTED; threadCtx->fwdCbCtx.isDirect = 0; From f08b701fbf42bead0efc87c44019b30421c06ece Mon Sep 17 00:00:00 2001 From: John Safranek Date: Tue, 22 Sep 2026 11:09:28 -0700 Subject: [PATCH 2/3] echoserver: end the session on a failed echo read wolfSSH_ChannelIdRead() returns a negative value only for a structural fault, never a hold to retry. app_echo_pump() now returns it so the session ends, and only a zero read marks the channel dry. --- examples/echoserver/echoserver.c | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/examples/echoserver/echoserver.c b/examples/echoserver/echoserver.c index 131677110..551e184bc 100644 --- a/examples/echoserver/echoserver.c +++ b/examples/echoserver/echoserver.c @@ -1183,9 +1183,14 @@ static int app_echo_pump(WOLFSSH* ssh, thread_ctx_t* threadCtx, int* wantWrite, if (app_staged(appCtx) == 0) { cnt = wolfSSH_ChannelIdRead(ssh, appCtx->channelId, appCtx->buffer, (word32)sizeof appCtx->buffer); - /* Only a zero read is dry; a negative one just stops the pump. */ - if (cnt <= 0) { - *dry = (cnt == 0); + /* Every negative here is structural -- WS_BAD_ARGUMENT, + * WS_INVALID_CHANID or WS_INVALID_STATE_E -- and never a hold + * to retry, so it ends the session rather than the pump. */ + if (cnt < 0) { + return cnt; + } + if (cnt == 0) { + *dry = 1; break; } #ifdef SHELL_DEBUG From f0f7131141d3b8579cc41d307914682f99d0e7f3 Mon Sep 17 00:00:00 2001 From: John Safranek Date: Tue, 22 Sep 2026 11:29:46 -0700 Subject: [PATCH 3/3] echoserver: pump channel data to nonblocking fds The pty and forward descriptors are nonblocking, and each pass pumps its channel's buffered data into them, holding what a write would not take. A blocked write no longer stalls the loop while the same peer waits on a window adjust only that loop can process. - the held tail puts the descriptor in the select() write set - a pump runs every pass, so buffered data needs no second data report - the forward gets a bounded final flush when the session ends - app_write_all() is left for the agent path alone --- examples/echoserver/echoserver.c | 213 +++++++++++++++++++++---------- 1 file changed, 147 insertions(+), 66 deletions(-) diff --git a/examples/echoserver/echoserver.c b/examples/echoserver/echoserver.c index 551e184bc..5d96648b9 100644 --- a/examples/echoserver/echoserver.c +++ b/examples/echoserver/echoserver.c @@ -189,6 +189,11 @@ typedef struct WS_AppCtx { * with 0 <= bufferOff <= bufferIdx. Both survive a worker pass. */ word32 bufferIdx; word32 bufferOff; + /* The other direction: channel data read for appFd and how many of + * those bytes appFd has taken, with 0 <= chanOff <= chanIdx. */ + byte chanBuffer[EXAMPLE_BUFFER_SZ]; + word32 chanIdx; + word32 chanOff; } WS_AppCtx; @@ -877,6 +882,14 @@ static int wsShellStartCb(WOLFSSH_CHANNEL* channel, void* ctx) ShellChildCleanup(threadCtx); return 1; } + /* A blocked write to the child would stall the session. */ + rc = fcntl(threadCtx->shellCtx.appFd, F_GETFL, 0); + if (rc < 0 || fcntl(threadCtx->shellCtx.appFd, F_SETFL, + rc | O_NONBLOCK) < 0) { + printf("fcntl failed: errno=%x\n", errno); + ShellChildCleanup(threadCtx); + return 1; + } /* Installed only now: the refusals above reap their own child. */ signal(SIGCHLD, ChildSig); @@ -1216,32 +1229,114 @@ static int app_echo_pump(WOLFSSH* ssh, thread_ctx_t* threadCtx, int* wantWrite, } -#if defined(WOLFSSH_SHELL) || defined(WOLFSSH_AGENT) || defined(WOLFSSH_FWD) +#if defined(WOLFSSH_SHELL) || defined(WOLFSSH_FWD) + +/* Bytes appCtx still owes its descriptor. */ +static word32 app_owed(const WS_AppCtx* appCtx) +{ + return appCtx->chanIdx - appCtx->chanOff; +} + -/* Write every byte to a blocking descriptor, retrying what an interrupted - * write left behind. Returns bufSz, or -1 */ -static int app_write_all(WS_SOCKET_T fd, const byte* buf, word32 bufSz, +/* Move the channel's buffered data to the nonblocking descriptor fd, holding + * what fd would not take until it turns writable. Runs every pass, since data + * already buffered inside the library draws no second data report. Returns + * 0, or negative to end the session. */ +static int app_pump_to_fd(WOLFSSH* ssh, WS_AppCtx* appCtx, WS_SOCKET_T fd, int isSocket) { - word32 off = 0; int cnt; #ifndef WOLFSSH_SHELL (void)isSocket; #endif - while (off < bufSz) { + for (;;) { + if (app_owed(appCtx) == 0) { + cnt = wolfSSH_ChannelIdRead(ssh, appCtx->channelId, + appCtx->chanBuffer, (word32)sizeof appCtx->chanBuffer); + /* Every negative is structural, as in app_echo_pump(). */ + if (cnt <= 0) { + return cnt; + } + #ifdef SHELL_DEBUG + buf_dump(appCtx->chanBuffer, cnt); + #endif + appCtx->chanIdx = (word32)cnt; + appCtx->chanOff = 0; + } + #ifdef WOLFSSH_SHELL if (!isSocket) { - cnt = (int)write(fd, buf + off, bufSz - off); + cnt = (int)write(fd, appCtx->chanBuffer + appCtx->chanOff, + app_owed(appCtx)); } else { - cnt = (int)send(fd, (const char*)buf + off, (int)(bufSz - off), 0); + cnt = (int)send(fd, (const char*)appCtx->chanBuffer + + appCtx->chanOff, (int)app_owed(appCtx), 0); } #else - cnt = (int)send(fd, (const char*)buf + off, (int)(bufSz - off), 0); + cnt = (int)send(fd, (const char*)appCtx->chanBuffer + appCtx->chanOff, + (int)app_owed(appCtx), 0); #endif + if (cnt > 0) { + appCtx->chanOff += (word32)cnt; + } + else if (cnt < 0 && SOCKET_ERRNO == SOCKET_EINTR) { + continue; + } + else if (cnt < 0 && SOCKET_ERRNO == SOCKET_EWOULDBLOCK) { + /* Held; fd joins the write set. */ + return 0; + } + else { + return WS_FATAL_ERROR; + } + } +} + +#endif /* WOLFSSH_SHELL || WOLFSSH_FWD */ + + +#ifdef WOLFSSH_FWD + +/* Seconds the forward may refuse more before the final flush gives up. */ +#define ES_FLUSH_TIMEOUT 1 + +/* The session is over, but the forward keeps what the channel still holds: + * hand it to fd, waiting on fd between attempts. */ +static void app_flush_to_fd(WOLFSSH* ssh, WS_AppCtx* appCtx, WS_SOCKET_T fd) +{ + fd_set writeFds; + struct timeval tv; + + while (app_pump_to_fd(ssh, appCtx, fd, 1) == 0 && app_owed(appCtx) > 0) { + FD_ZERO(&writeFds); + FD_SET(fd, &writeFds); + tv.tv_sec = ES_FLUSH_TIMEOUT; + tv.tv_usec = 0; + if (select((int)fd + 1, NULL, &writeFds, NULL, &tv) <= 0) { + break; + } + } +} + +#endif /* WOLFSSH_FWD */ + + +#ifdef WOLFSSH_AGENT + +/* Write every byte to a blocking socket, retrying what an interrupted send + * left behind. Returns bufSz, or -1 */ +static int app_write_all(WS_SOCKET_T fd, const byte* buf, word32 bufSz) +{ + word32 off = 0; + int cnt; + + while (off < bufSz) { + cnt = (int)send(fd, (const char*)buf + off, (int)(bufSz - off), 0); + if (cnt > 0) { off += (word32)cnt; } @@ -1256,7 +1351,7 @@ static int app_write_all(WS_SOCKET_T fd, const byte* buf, word32 bufSz, return (int)bufSz; } -#endif /* WOLFSSH_SHELL || WOLFSSH_AGENT || WOLFSSH_FWD */ +#endif /* WOLFSSH_AGENT */ static int ssh_worker(thread_ctx_t* threadCtx) @@ -1380,6 +1475,14 @@ static int ssh_worker(thread_ctx_t* threadCtx) if (threadCtx->shellCtx.appFd > maxFd) maxFd = threadCtx->shellCtx.appFd; } + /* ... and in the write set while it owes channel data */ + if (threadCtx->shellCtx.state == APP_STATE_CONNECTED + && threadCtx->shellCtx.appFd >= 0 + && app_owed(&threadCtx->shellCtx) > 0) { + FD_SET(threadCtx->shellCtx.appFd, &writeFds); + if (threadCtx->shellCtx.appFd > maxFd) + maxFd = threadCtx->shellCtx.appFd; + } #endif /* WOLFSSH_SHELL */ #ifdef WOLFSSH_AGENT /* The poll above creates this listener mid-loop; re-read it @@ -1414,10 +1517,16 @@ static int ssh_worker(thread_ctx_t* threadCtx) if (fwdFd > maxFd) maxFd = fwdFd; } + if (fwdFd >= 0 + && threadCtx->fwdCtx.state == APP_STATE_CONNECTED + && app_owed(&threadCtx->fwdCtx) > 0) { + FD_SET(fwdFd, &writeFds); + if (fwdFd > maxFd) + maxFd = fwdFd; + } #endif /* WOLFSSH_FWD */ - rc = select((int)maxFd + 1, &readFds, - wantWrite ? &writeFds : NULL, NULL, NULL); + rc = select((int)maxFd + 1, &readFds, &writeFds, NULL, NULL); if (rc == -1) { break; } @@ -1500,37 +1609,11 @@ static int ssh_worker(thread_ctx_t* threadCtx) * report is never raised again, so drain on both or the * buffered bytes sit there and the peer waits forever. * wolfSSH_ChannelIdRead() has no isKeying gate; the window - * credit it owes is parked until the rekey finishes. */ + * credit it owes is parked until the rekey finishes. The + * shell and forward channels are served by the pumps + * below, which run on every pass rather than on a data + * report. */ if (rc == WS_CHAN_RXD || rc == WS_REKEYING) { - #ifdef WOLFSSH_SHELL - /* Echo mode is served by app_echo_pump(), which runs - * on every pass rather than on a data report. */ - if (!echoOnly - && threadCtx->shellCtx.state - == APP_STATE_CONNECTED - && lastChannel - == threadCtx->shellCtx.channelId) { - cnt_r = wolfSSH_ChannelIdRead(ssh, - threadCtx->shellCtx.channelId, - threadCtx->channelBuffer, - sizeof threadCtx->channelBuffer); - if (cnt_r <= 0) { - /* Nothing was buffered. Only an actual data - * report makes that a failure. */ - if (rc == WS_REKEYING && cnt_r == 0) - continue; - break; - } - #ifdef SHELL_DEBUG - buf_dump(threadCtx->channelBuffer, cnt_r); - #endif - if (app_write_all(threadCtx->shellCtx.appFd, - threadCtx->channelBuffer, - (word32)cnt_r, 0) < 0) { - break; - } - } - #endif /* WOLFSSH_SHELL */ #ifdef WOLFSSH_AGENT if (lastChannel == agentChannelId) { cnt_r = wolfSSH_ChannelIdRead(ssh, agentChannelId, @@ -1547,30 +1630,7 @@ static int ssh_worker(thread_ctx_t* threadCtx) buf_dump(threadCtx->channelBuffer, cnt_r); #endif if (app_write_all(agentFd, threadCtx->channelBuffer, - (word32)cnt_r, 1) < 0) - break; - } - #endif - #ifdef WOLFSSH_FWD - if (threadCtx->fwdCtx.state == APP_STATE_CONNECTED && - lastChannel == threadCtx->fwdCtx.channelId) { - - cnt_r = wolfSSH_ChannelIdRead(ssh, - threadCtx->fwdCtx.channelId, - threadCtx->channelBuffer, - sizeof threadCtx->channelBuffer); - if (cnt_r <= 0) { - /* Nothing was buffered. Only an actual data - * report makes that a failure. */ - if (rc == WS_REKEYING && cnt_r == 0) - continue; - break; - } - #ifdef SHELL_DEBUG - buf_dump(threadCtx->channelBuffer, cnt_r); - #endif - if (app_write_all(fwdFd, threadCtx->channelBuffer, - (word32)cnt_r, 1) < 0) + (word32)cnt_r) < 0) break; } #endif @@ -1672,6 +1732,13 @@ static int ssh_worker(thread_ctx_t* threadCtx) &wantWrite) < 0) { break; } + #ifdef WOLFSSH_SHELL + if (threadCtx->shellCtx.appFd >= 0 + && app_pump_to_fd(ssh, &threadCtx->shellCtx, + threadCtx->shellCtx.appFd, 0) < 0) { + break; + } + #endif } else if (app_echo_pump(ssh, threadCtx, &wantWrite, &echoDry) < 0) { @@ -1803,6 +1870,9 @@ static int ssh_worker(thread_ctx_t* threadCtx) threadCtx->fwdCtx.channelId, &wantWrite) < 0) { break; } + if (app_pump_to_fd(ssh, &threadCtx->fwdCtx, fwdFd, 1) < 0) { + break; + } } if (threadCtx->fwdCtx.state == APP_STATE_LISTEN && threadCtx->fwdCtx.listenFd >= 0) { @@ -1823,6 +1893,7 @@ static int ssh_worker(thread_ctx_t* threadCtx) const char* out = NULL; char addr[200]; + tcp_set_nonblocking(&fwdFd); threadCtx->fwdCtx.state = APP_STATE_CONNECT; threadCtx->fwdCtx.appFd = fwdFd; originAddrSz = sizeof originAddr; @@ -1865,6 +1936,8 @@ static int ssh_worker(thread_ctx_t* threadCtx) if (newChannel != NULL) { threadCtx->fwdCtx.bufferIdx = 0; threadCtx->fwdCtx.bufferOff = 0; + threadCtx->fwdCtx.chanIdx = 0; + threadCtx->fwdCtx.chanOff = 0; threadCtx->fwdCtx.state = APP_STATE_CONNECTED; } } @@ -1873,8 +1946,11 @@ static int ssh_worker(thread_ctx_t* threadCtx) threadCtx->fwdCbCtx.hostPort); if (fwdFd > 0) { + tcp_set_nonblocking(&fwdFd); threadCtx->fwdCtx.bufferIdx = 0; threadCtx->fwdCtx.bufferOff = 0; + threadCtx->fwdCtx.chanIdx = 0; + threadCtx->fwdCtx.chanOff = 0; threadCtx->fwdCtx.appFd = fwdFd; threadCtx->fwdCtx.state = APP_STATE_CONNECTED; threadCtx->fwdCbCtx.isDirect = 0; @@ -1882,6 +1958,11 @@ static int ssh_worker(thread_ctx_t* threadCtx) } #endif /* WOLFSSH_FWD */ } +#ifdef WOLFSSH_FWD + if (fwdFd >= 0 && threadCtx->fwdCtx.state == APP_STATE_CONNECTED) { + app_flush_to_fd(ssh, &threadCtx->fwdCtx, fwdFd); + } +#endif #ifdef WOLFSSH_SHELL ShellChildCleanup(threadCtx); #endif