diff --git a/examples/echoserver/echoserver.c b/examples/echoserver/echoserver.c index 4dde815b6..5d96648b9 100644 --- a/examples/echoserver/echoserver.c +++ b/examples/echoserver/echoserver.c @@ -185,6 +185,15 @@ 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; + /* 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; @@ -236,9 +245,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; @@ -876,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); @@ -1113,6 +1127,233 @@ 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); + /* 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 + 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_FWD) + +/* Bytes appCtx still owes its descriptor. */ +static word32 app_owed(const WS_AppCtx* appCtx) +{ + return appCtx->chanIdx - appCtx->chanOff; +} + + +/* 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) +{ + int cnt; + +#ifndef WOLFSSH_SHELL + (void)isSocket; +#endif + + 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, appCtx->chanBuffer + appCtx->chanOff, + app_owed(appCtx)); + } + else { + cnt = (int)send(fd, (const char*)appCtx->chanBuffer + + appCtx->chanOff, (int)app_owed(appCtx), 0); + } +#else + 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; + } + else if (cnt < 0 && SOCKET_ERRNO == SOCKET_EINTR) { + continue; + } + else { + return -1; + } + } + + return (int)bufSz; +} + +#endif /* WOLFSSH_AGENT */ + + static int ssh_worker(thread_ctx_t* threadCtx) { WOLFSSH* ssh; @@ -1122,9 +1363,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 +1425,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 +1435,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,13 +1465,24 @@ 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; } + /* ... 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 @@ -1247,7 +1494,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,15 +1511,22 @@ 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; } + 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; } @@ -1324,39 +1579,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 @@ -1382,54 +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) { - if (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 - #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) - break; - } #ifdef WOLFSSH_AGENT if (lastChannel == agentChannelId) { cnt_r = wolfSSH_ChannelIdRead(ssh, agentChannelId, @@ -1445,33 +1629,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) - 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 - cnt_w = (int)send(fwdFd, threadCtx->channelBuffer, - cnt_r, 0); - if (cnt_w <= 0) + if (app_write_all(agentFd, threadCtx->channelBuffer, + (word32)cnt_r) < 0) break; } #endif @@ -1556,17 +1715,36 @@ 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; + } + #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) { + break; + } + } #ifdef WOLFSSH_AGENT if (agentFd >= 0 && threadCtx->agentCtx.state == APP_STATE_CONNECTED) { @@ -1608,13 +1786,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 +1809,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 +1824,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 +1862,16 @@ 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 (app_pump_to_fd(ssh, &threadCtx->fwdCtx, fwdFd, 1) < 0) { + break; } } if (threadCtx->fwdCtx.state == APP_STATE_LISTEN @@ -1722,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; @@ -1762,6 +1934,10 @@ 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.chanIdx = 0; + threadCtx->fwdCtx.chanOff = 0; threadCtx->fwdCtx.state = APP_STATE_CONNECTED; } } @@ -1770,6 +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; @@ -1777,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