ftpsrvr.c

/* Synchronet FTP server */

/****************************************************************************
 * @format.tab-size 4		(Plain Text/Source Code File Header)			*
 * @format.use-tabs true	(see http://www.synchro.net/ptsc_hdr.html)		*
 *																			*
 * Copyright Rob Swindell - http://www.synchro.net/copyright.html			*
 *																			*
 * This program is free software; you can redistribute it and/or			*
 * modify it under the terms of the GNU General Public License				*
 * as published by the Free Software Foundation; either version 2			*
 * of the License, or (at your option) any later version.					*
 * See the GNU General Public License for more details: gpl.txt or			*
 * http://www.fsf.org/copyleft/gpl.html										*
 *																			*
 * For Synchronet coding style and modification guidelines, see				*
 * http://www.synchro.net/source.html										*
 *																			*
 * Note: If this box doesn't appear square, then you need to fix your tabs.	*
 ****************************************************************************/

/* ANSI C Library headers */
#include <stdio.h>
#include <stdlib.h>         /* ltoa in GNU C lib */
#include <stdarg.h>         /* va_list, varargs */
#include <string.h>         /* strrchr */
#include <fcntl.h>          /* O_WRONLY, O_RDONLY, etc. */
#include <errno.h>          /* EACCES */
#include <ctype.h>          /* toupper */
#include <sys/types.h>
#include <sys/stat.h>

/* Synchronet-specific headers */
#undef SBBS /* this shouldn't be defined unless building sbbs.dll/libsbbs.so */
#include "sbbs.h"
#include "text.h"           /* TOTAL_TEXT */
#include "ftpsrvr.h"
#include "filedat.h"
#include "telnet.h"
#include "multisock.h"
#include "ssl.h"
#include "cryptlib.h"
#include "xpprintf.h"       // vasprintf
#include "md5.h"
#include "sauce.h"
#include "git_branch.h"
#include "git_hash.h"

/* Constants */

#define FTP_SERVER              "Synchronet FTP Server"
static const char* server_abbrev = "ftp";

#define ANONYMOUS               "anonymous"

#define BBS_VIRTUAL_PATH        "bbs:/""/"  /* this is actually bbs:<slash><slash> */
#define LOCAL_FSYS_DIR          "local:"
#define BBS_FSYS_DIR            "bbs:"
#define BBS_HIDDEN_ALIAS        "hidden"

#define TIMEOUT_THREAD_WAIT     60      /* Seconds */

#define TIMEOUT_SOCKET_LISTEN   30      /* Seconds */

#define XFER_REPORT_INTERVAL    60      /* Seconds */

#define INDEX_FNAME_LEN         15

#define NAME_LEN                15      /* User name length for listings */

#define MLSX_TYPE   (1 << 0)
#define MLSX_PERM   (1 << 1)
#define MLSX_SIZE   (1 << 2)
#define MLSX_MODIFY (1 << 3)
#define MLSX_OWNER  (1 << 4)
#define MLSX_UNIQUE (1 << 5)
#define MLSX_CREATE (1 << 6)

static ftp_startup_t*     startup = NULL;
static scfg_t             scfg;
static struct mqtt        mqtt;
static struct xpms_set *  ftp_set = NULL;
static protected_uint32_t active_clients;
static protected_uint32_t thread_count;
static volatile uint32_t  client_highwater = 0;
static volatile time_t    uptime = 0;
static volatile ulong     served = 0;
static bool               terminate_server = FALSE;
static char *             text[TOTAL_TEXT];
static str_list_t         pause_semfiles;
static str_list_t         recycle_semfiles;
static str_list_t         shutdown_semfiles;
static link_list_t        current_connections;

#ifdef SOCKET_DEBUG
static BYTE               socket_debug[0x10000] = {0};

	#define SOCKET_DEBUG_CTRL       (1 << 0)  /* 0x01 */
	#define SOCKET_DEBUG_SEND       (1 << 1)  /* 0x02 */
	#define SOCKET_DEBUG_READLINE   (1 << 2)  /* 0x04 */
	#define SOCKET_DEBUG_ACCEPT     (1 << 3)  /* 0x08 */
	#define SOCKET_DEBUG_SENDTHREAD (1 << 4)  /* 0x10 */
	#define SOCKET_DEBUG_TERMINATE  (1 << 5)  /* 0x20 */
	#define SOCKET_DEBUG_RECV_CHAR  (1 << 6)  /* 0x40 */
	#define SOCKET_DEBUG_FILEXFER   (1 << 7)  /* 0x80 */
#endif

char* genvpath(int lib, int dir, char* str);

typedef struct {
	SOCKET socket;
	union xp_sockaddr client_addr;
	socklen_t client_addr_len;
} ftp_t;


static const char *ftp_mon[] = {"Jan", "Feb", "Mar", "Apr", "May", "Jun"
	                            , "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"};

BOOL direxist(char *dir)
{
	if (access(dir, 0) == 0)
		return TRUE;
	else
		return FALSE;
}

BOOL dir_op(scfg_t* cfg, user_t* user, client_t* client, uint dirnum)
{
	return user_is_dirop(cfg, dirnum, user, client);
}

static int lputs(int level, const char* str)
{
	mqtt_lputs(&mqtt, TOPIC_SERVER, level, str);
	if (level <= LOG_ERR) {
		char errmsg[1024];
		SAFEPRINTF2(errmsg, "%-4s %s", server_abbrev, str);
		errorlog(&scfg, &mqtt, level, startup == NULL ? NULL : startup->host_name, errmsg);
		if (startup != NULL && startup->errormsg != NULL)
			startup->errormsg(startup->cbdata, level, errmsg);
	}

	if (startup == NULL || startup->lputs == NULL || str == NULL || level > startup->log_level)
		return 0;

#if defined(_WIN32)
	if (IsBadCodePtr((FARPROC)startup->lputs))
		return 0;
#endif

	return startup->lputs(startup->cbdata, level, str);
}

#if defined(__GNUC__)   // Catch printf-format errors with lprintf
static int lprintf(int level, const char *fmt, ...) __attribute__ ((format (printf, 2, 3)));
#endif
static int lprintf(int level, const char *fmt, ...)
{
	va_list argptr;
	char    sbuf[1024];

	va_start(argptr, fmt);
	vsnprintf(sbuf, sizeof(sbuf), fmt, argptr);
	sbuf[sizeof(sbuf) - 1] = 0;
	va_end(argptr);

	return lputs(level, sbuf);
}

#ifdef _WINSOCKAPI_

static WSADATA WSAData;
#define SOCKLIB_DESC WSAData.szDescription

static BOOL    WSAInitialized = FALSE;

static BOOL winsock_startup(void)
{
	int status;                 /* Status Code */

	if ((status = WSAStartup(MAKEWORD(1, 1), &WSAData)) == 0) {
		lprintf(LOG_DEBUG, "%s %s", WSAData.szDescription, WSAData.szSystemStatus);
		WSAInitialized = TRUE;
		return TRUE;
	}

	lprintf(LOG_CRIT, "!WinSock startup ERROR %d", status);
	return FALSE;
}

#else /* No WINSOCK */

#define winsock_startup()   (TRUE)
#define SOCKLIB_DESC        NULL

#endif

static char* server_host_name(void)
{
	return startup->host_name[0] ? startup->host_name : scfg.sys_inetaddr;
}

static void set_state(enum server_state state)
{
	static int curr_state;

	if (state == curr_state)
		return;

	if (startup != NULL) {
		if (startup->set_state != NULL)
			startup->set_state(startup->cbdata, state);
		mqtt_server_state(&mqtt, state);
	}
	curr_state = state;
}

static void update_clients(void)
{
	if (startup != NULL) {
		uint32_t count = protected_uint32_value(active_clients);
		if (startup->clients != NULL)
			startup->clients(startup->cbdata, count);
	}
}

static void client_on(SOCKET sock, client_t* client, BOOL update)
{
	if (!update)
		listAddNodeData(&current_connections, client->addr, strlen(client->addr) + 1, sock, LAST_NODE);
	if (startup != NULL && startup->client_on != NULL)
		startup->client_on(startup->cbdata, TRUE, sock, client, update);
	mqtt_client_on(&mqtt, TRUE, sock, client, update);
}

static void client_off(SOCKET sock)
{
	listRemoveTaggedNode(&current_connections, sock, /* free_data */ TRUE);
	if (startup != NULL && startup->client_on != NULL)
		startup->client_on(startup->cbdata, FALSE, sock, NULL, FALSE);
	mqtt_client_on(&mqtt, FALSE, sock, NULL, FALSE);
}

static void thread_up(BOOL setuid)
{
	if (startup != NULL) {
		if (startup->thread_up != NULL)
			startup->thread_up(startup->cbdata, TRUE, setuid);
	}
}

static int32_t thread_down(void)
{
	int32_t count = protected_uint32_adjust_fetch(&thread_count, -1);
	if (startup != NULL) {
		if (startup->thread_up != NULL)
			startup->thread_up(startup->cbdata, FALSE, FALSE);
	}
	return count;
}

static void ftp_open_socket_cb(SOCKET sock, void *cbdata)
{
	char error[256];

	if (startup != NULL && startup->socket_open != NULL)
		startup->socket_open(startup->cbdata, TRUE);
	if (set_socket_options(&scfg, sock, "FTP", error, sizeof(error)))
		lprintf(LOG_ERR, "%04d !ERROR %s", sock, error);
}

static void ftp_close_socket_cb(SOCKET sock, void *cbdata)
{
	if (startup != NULL && startup->socket_open != NULL)
		startup->socket_open(startup->cbdata, FALSE);
}

static SOCKET ftp_open_socket(int domain, int type)
{
	SOCKET sock;

	sock = socket(domain, type, IPPROTO_IP);
	if (sock != INVALID_SOCKET)
		ftp_open_socket_cb(sock, NULL);
	return sock;
}

#ifdef __BORLANDC__
#pragma argsused
#endif
static int ftp_close_socket(SOCKET* sock, CRYPT_SESSION *sess, int line)
{
	int result;

	if (*sess != -1) {
		destroy_session(lprintf, *sess);
		*sess = -1;
	}

	if ((*sock) == INVALID_SOCKET) {
		lprintf(LOG_WARNING, "0000 !INVALID_SOCKET in close_socket from line %u", line);
		return -1;
	}

	shutdown(*sock, SHUT_RDWR);  /* required on Unix */

	result = closesocket(*sock);
	if (startup != NULL && startup->socket_open != NULL)
		startup->socket_open(startup->cbdata, FALSE);

	if (result != 0) {
		if (SOCKET_ERRNO != ENOTSOCK)
			lprintf(LOG_WARNING, "%04d !ERROR %d closing socket from line %u", *sock, SOCKET_ERRNO, line);
	}
	*sock = INVALID_SOCKET;

	return result;
}

#define GCES(status, sock, session, estr, action) do {                  \
			int GCES_level;                                                     \
			get_crypt_error_string(status, session, &estr, action, &GCES_level); \
			if (estr != NULL) {                                                 \
				lprintf(GCES_level, "%04d TLS %s", sock, estr);                 \
				free_crypt_attrstr(estr);                                       \
				estr = NULL;                                                      \
			}                                                                   \
} while (0)


#if defined(__GNUC__)   // Catch printf-format errors with sockprintf
static int sockprintf(SOCKET sock, CRYPT_SESSION sess, char *fmt, ...) __attribute__ ((format (printf, 3, 4)));
#endif
static int sockprintf(SOCKET sock, CRYPT_SESSION sess, char *fmt, ...)
{
	int     len;
	int     maxlen;
	int     result;
	va_list argptr;
	char    sbuf[1024];
	char *  estr;

	va_start(argptr, fmt);
	len = vsnprintf(sbuf, maxlen = sizeof(sbuf) - 2, fmt, argptr);
	va_end(argptr);

	if (len < 0 || len > maxlen) /* format error or output truncated */
		len = maxlen;
	if (startup != NULL && startup->options & FTP_OPT_DEBUG_TX)
		lprintf(LOG_DEBUG, "%04d TX%s: %.*s", sock, sess != -1 ? "S" : "", len, sbuf);
	memcpy(sbuf + len, "\r\n", 2);
	len += 2;

	if (sock == INVALID_SOCKET) {
		lprintf(LOG_WARNING, "!INVALID SOCKET in call to sockprintf");
		return 0;
	}

	/* Check socket for writability */
	if (!socket_writable(sock, 300000)) {
		lprintf(LOG_WARNING, "%04d !WARNING socket not ready for write", sock);
		return 0;
	}

	if (sess != -1) {
		int tls_sent;
		int sent = 0;
		while (sent < len) {
			result = cryptPushData(sess, sbuf + sent, len - sent, &tls_sent);
			if (result == CRYPT_OK)
				sent += tls_sent;
			else {
				GCES(result, sock, sess, estr, "sending data");
				if (result != CRYPT_ERROR_TIMEOUT)
					return 0;
			}
			result = cryptFlushData(sess);
			if (result != CRYPT_OK) {
				GCES(result, sock, sess, estr, "flushing data");
				return 0;
			}
		}
	}
	else {
		while ((result = sendsocket(sock, sbuf, len)) != len) {
			if (result == SOCKET_ERROR) {
				if (SOCKET_ERRNO == EWOULDBLOCK) {
					YIELD();
					continue;
				}
				if (SOCKET_ERRNO == ECONNRESET)
					lprintf(LOG_WARNING, "%04d Connection reset by peer on send", sock);
				else if (SOCKET_ERRNO == ECONNABORTED)
					lprintf(LOG_WARNING, "%04d Connection aborted by peer on send", sock);
				else
					lprintf(LOG_WARNING, "%04d !ERROR %d sending", sock, SOCKET_ERRNO);
				return 0;
			}
			lprintf(LOG_WARNING, "%04d !ERROR: short send: %u instead of %u", sock, result, len);
		}
	}
	return len;
}

void recverror(SOCKET socket, int rd, int line)
{
	if (rd == 0)
		lprintf(LOG_NOTICE, "%04d Socket closed by peer on receive (line %u)"
		        , socket, line);
	else if (rd == SOCKET_ERROR) {
		if (SOCKET_ERRNO == ECONNRESET)
			lprintf(LOG_NOTICE, "%04d Connection reset by peer on receive (line %u)"
			        , socket, line);
		else if (SOCKET_ERRNO == ECONNABORTED)
			lprintf(LOG_NOTICE, "%04d Connection aborted by peer on receive (line %u)"
			        , socket, line);
		else
			lprintf(LOG_NOTICE, "%04d !ERROR %d receiving on socket (line %u)"
			        , socket, SOCKET_ERRNO, line);
	} else
		lprintf(LOG_WARNING, "%04d !ERROR: recv on socket returned unexpected value: %d (line %u)"
		        , socket, rd, line);
}

static int sock_recvbyte(SOCKET sock, CRYPT_SESSION sess, char *buf, time_t *lastactive)
{
	int   len = 0;
	int   ret;
	int   i;
	char *estr;
	BOOL  first = TRUE;

	if (ftp_set == NULL || terminate_server) {
		sockprintf(sock, sess, "421 Server downed, aborting.");
		lprintf(LOG_WARNING, "%04d Server downed, aborting", sock);
		return 0;
	}
	if (sess > -1) {
		/* Try a read with no timeout first. */
		if ((ret = cryptSetAttribute(sess, CRYPT_OPTION_NET_READTIMEOUT, 0)) != CRYPT_OK)
			GCES(ret, sock, sess, estr, "setting read timeout");
		while (1) {
			ret = cryptPopData(sess, buf, 1, &len);
			/* Successive reads will be with the full timeout after a socket_readable() */
			cryptSetAttribute(sess, CRYPT_OPTION_NET_READTIMEOUT, startup->max_inactivity);
			switch (ret) {
				case CRYPT_OK:
					break;
				case CRYPT_ERROR_TIMEOUT:
					if (!first) {
						GCES(ret, sock, sess, estr, "popping data");
						return -1;
					}
					break;
				case CRYPT_ERROR_COMPLETE:
					return 0;
				default:
					GCES(ret, sock, sess, estr, "popping data");
					if (ret < -1)
						return ret;
					return -2;
			}
			first = FALSE;
			if (len)
				return len;

			if ((time(NULL) - (*lastactive)) > startup->max_inactivity) {
				lprintf(LOG_WARNING, "%04d Disconnecting due to to inactivity", sock);
				sockprintf(sock, sess, "421 Disconnecting due to inactivity (%u seconds)."
				           , startup->max_inactivity);
				return 0;
			}

			if (!socket_readable(sock, startup->max_inactivity * 1000)) {
				if ((time(NULL) - (*lastactive)) > startup->max_inactivity) {
					lprintf(LOG_WARNING, "%04d Disconnecting due to to inactivity", sock);
					sockprintf(sock, sess, "421 Disconnecting due to inactivity (%u seconds)."
					           , startup->max_inactivity);
					return 0;
				}
			}
		}
	}
	else {
		while (1) {
			if (!socket_readable(sock, startup->max_inactivity * 1000)) {
				if ((time(NULL) - (*lastactive)) > startup->max_inactivity) {
					lprintf(LOG_WARNING, "%04d Disconnecting due to to inactivity", sock);
					sockprintf(sock, sess, "421 Disconnecting due to inactivity (%u seconds)."
					           , startup->max_inactivity);
					return 0;
				}
				continue;
			}
	#ifdef SOCKET_DEBUG_RECV_CHAR
			socket_debug[sock] |= SOCKET_DEBUG_RECV_CHAR;
	#endif
			i = recv(sock, buf, 1, 0);
	#ifdef SOCKET_DEBUG_RECV_CHAR
			socket_debug[sock] &= ~SOCKET_DEBUG_RECV_CHAR;
	#endif
			return i;
		}
	}
}

int sockreadline(SOCKET socket, CRYPT_SESSION sess, char* buf, int len, time_t* lastactive)
{
	char ch;
	int  i, rd = 0;

	buf[0] = 0;

	if (socket == INVALID_SOCKET) {
		lprintf(LOG_WARNING, "INVALID SOCKET in call to sockreadline");
		return 0;
	}

	while (rd < len - 1) {
		i = sock_recvbyte(socket, sess, &ch, lastactive);

		if (i < 1) {
			if (sess != -1)
				recverror(socket, i, __LINE__);
			return i;
		}
		if (ch == '\n' /* && rd>=1 */) { /* Mar-9-2003: terminate on sole LF */
			break;
		}
		buf[rd++] = ch;
	}
	if (rd > 0 && buf[rd - 1] == '\r')
		buf[rd - 1] = 0;
	else
		buf[rd] = 0;

	return rd;
}

void ftp_terminate(void)
{
	lprintf(LOG_INFO, "FTP Server terminate");
	terminate_server = TRUE;
}

bool ftp_remove(SOCKET sock, int line, const char* fname, const char* username, int err_level)
{
	int ret = 0;

	if (fexist(fname) && (ret = remove(fname)) != 0) {
		if (fexist(fname)) { // In case there was a race condition (other host deleted file first)
			char error[256];
			lprintf(err_level, "%04d <%s> !ERROR %d (%s) (line %d) removing file: %s"
			        , sock, username, errno, safe_strerror(errno, error, sizeof error), line, fname);
		}
	}
	return ret == 0;
}

typedef struct {
	SOCKET ctrl_sock;
	CRYPT_SESSION ctrl_sess;
	SOCKET* data_sock;
	CRYPT_SESSION* data_sess;
	BOOL* inprogress;
	BOOL* aborted;
	BOOL delfile;
	BOOL tmpfile;
	BOOL credits;
	BOOL append;
	off_t filepos;
	char filename[MAX_PATH + 1];
	time_t* lastactive;
	user_t* user;
	client_t* client;
	int dir;
	char* desc;
} xfer_t;

static void send_thread(void* arg)
{
	char              buf[8192];
	char              str[256];
	char              errstr[256];
	char              tmp[128];
	char              username[128];
	char              host_ip[INET6_ADDRSTRLEN];
	int               i;
	int               rd;
	int               wr;
	long              mod;
	uint64_t          l;
	off_t             total = 0;
	off_t             last_total = 0;
	ulong             dur;
	ulong             cps;
	off_t             length;
	BOOL              error = FALSE;
	FILE*             fp;
	file_t            f;
	xfer_t            xfer;
	time_t            now;
	time_t            start;
	time_t            last_report;
	user_t            uploader;
	union xp_sockaddr addr;
	socklen_t         addr_len;
	char *            estr;

	xfer = *(xfer_t*)arg;
	free(arg);

	SetThreadName("sbbs/ftpSend");
	thread_up(TRUE /* setuid */);

	length = flength(xfer.filename);

	if (length < 1) {
		if (xfer.tmpfile) {
			if (!(startup->options & FTP_OPT_KEEP_TEMP_FILES))
				ftp_remove(xfer.ctrl_sock, __LINE__, xfer.filename, xfer.user->alias, LOG_ERR);
			sockprintf(xfer.ctrl_sock, xfer.ctrl_sess, "450 No files");
		} else {
			lprintf(LOG_WARNING, "%04d <%s> !DATA cannot send file (%s) with size of %" PRIdOFF " bytes"
			        , xfer.ctrl_sock, xfer.user->alias, xfer.filename, length);
			sockprintf(xfer.ctrl_sock, xfer.ctrl_sess, "450 Invalid file size: %" PRIdOFF, length);
		}
		ftp_close_socket(xfer.data_sock, xfer.data_sess, __LINE__);
		*xfer.inprogress = FALSE;
		thread_down();
		return;
	}

	if ((fp = fnopen(NULL, xfer.filename, O_RDONLY | O_BINARY)) == NULL  /* non-shareable open failed */
	    && (fp = fopen(xfer.filename, "rb")) == NULL) {              /* shareable open failed */
		lprintf(LOG_ERR, "%04d <%s> !DATA ERROR %d (%s) line %d opening %s"
		        , xfer.ctrl_sock, xfer.user->alias, errno, safe_strerror(errno, errstr, sizeof errstr), __LINE__, xfer.filename);
		sockprintf(xfer.ctrl_sock, xfer.ctrl_sess, "450 ERROR %d (%s) opening %s", errno, safe_strerror(errno, errstr, sizeof errstr), xfer.filename);
		if (xfer.tmpfile && !(startup->options & FTP_OPT_KEEP_TEMP_FILES))
			ftp_remove(xfer.ctrl_sock, __LINE__, xfer.filename, xfer.user->alias, LOG_ERR);
		ftp_close_socket(xfer.data_sock, xfer.data_sess, __LINE__);
		*xfer.inprogress = FALSE;
		thread_down();
		return;
	}

#ifdef SOCKET_DEBUG_SENDTHREAD
	socket_debug[xfer.ctrl_sock] |= SOCKET_DEBUG_SENDTHREAD;
#endif

	*xfer.aborted = FALSE;
	if (xfer.filepos < 0)
		xfer.filepos = 0;
	if (startup->options & FTP_OPT_DEBUG_DATA || xfer.filepos)
		lprintf(LOG_DEBUG, "%04d <%s> DATA socket %d sending %s from offset %" PRIdOFF
		        , xfer.ctrl_sock, xfer.user->alias, *xfer.data_sock, xfer.filename, xfer.filepos);

	fseeko(fp, xfer.filepos, SEEK_SET);
	last_report = start = time(NULL);
	while ((xfer.filepos + total) < length) {

		now = time(NULL);

		/* Periodic progress report */
		if (total && now >= last_report + XFER_REPORT_INTERVAL) {
			if (xfer.filepos)
				sprintf(str, " from offset %" PRIdOFF, xfer.filepos);
			else
				str[0] = 0;
			lprintf(LOG_INFO, "%04d <%s> DATA Sent %" PRIdOFF " bytes (%" PRIdOFF " total) of %s (%lu cps)%s"
			        , xfer.ctrl_sock, xfer.user->alias, total, length, xfer.filename
			        , (ulong)((total - last_total) / (now - last_report))
			        , str);
			last_total = total;
			last_report = now;
		}

		if (*xfer.aborted == TRUE) {
			lprintf(LOG_WARNING, "%04d <%s> !DATA Transfer aborted", xfer.ctrl_sock, xfer.user->alias);
			sockprintf(xfer.ctrl_sock, xfer.ctrl_sess, "426 Transfer aborted.");
			error = TRUE;
			break;
		}
		if (ftp_set == NULL || terminate_server) {
			lprintf(LOG_WARNING, "%04d <%s> !DATA Transfer locally aborted", xfer.ctrl_sock, xfer.user->alias);
			sockprintf(xfer.ctrl_sock, xfer.ctrl_sess, "426 Transfer locally aborted.");
			error = TRUE;
			break;
		}

		/* Check socket for writability */
		if (!socket_writable(*xfer.data_sock, 1000))
			continue;

		fseeko(fp, xfer.filepos + total, SEEK_SET);
		rd = fread(buf, sizeof(char), sizeof(buf), fp);
		if (rd < 1) /* EOF or READ error */
			break;

#ifdef SOCKET_DEBUG_SEND
		socket_debug[xfer.ctrl_sock] |= SOCKET_DEBUG_SEND;
#endif
		if (*xfer.data_sess != -1) {
			int status = cryptPushData(*xfer.data_sess, buf, rd, &wr);
			if (status != CRYPT_OK) {
				GCES(status, *xfer.data_sock, *xfer.data_sess, estr, "pushing data");
				wr = -1;
			}
			else {
				status = cryptFlushData(*xfer.data_sess);
				if (status != CRYPT_OK) {
					GCES(status, *xfer.data_sock, *xfer.data_sess, estr, "flushing data");
					wr = -1;
				}
			}
		}
		else
			wr = sendsocket(*xfer.data_sock, buf, rd);
#ifdef SOCKET_DEBUG_SEND
		socket_debug[xfer.ctrl_sock] &= ~SOCKET_DEBUG_SEND;
#endif
		if (wr < 1) {
			if (wr == SOCKET_ERROR) {
				if (SOCKET_ERRNO == EWOULDBLOCK) {
					/*lprintf(LOG_WARNING,"%04d DATA send would block, retrying",xfer.ctrl_sock);*/
					YIELD();
					continue;
				}
				else if (SOCKET_ERRNO == ECONNRESET)
					lprintf(LOG_WARNING, "%04d <%s> DATA Connection reset by peer, sending on socket %d"
					        , xfer.ctrl_sock, xfer.user->alias, *xfer.data_sock);
				else if (SOCKET_ERRNO == ECONNABORTED)
					lprintf(LOG_WARNING, "%04d <%s> DATA Connection aborted by peer, sending on socket %d"
					        , xfer.ctrl_sock, xfer.user->alias, *xfer.data_sock);
				else
					lprintf(LOG_WARNING, "%04d <%s> !DATA ERROR %d sending on data socket %d"
					        , xfer.ctrl_sock, xfer.user->alias, SOCKET_ERRNO, *xfer.data_sock);
				/* Send NAK */
				sockprintf(xfer.ctrl_sock, xfer.ctrl_sess, "426 Error %d sending on DATA channel"
				           , SOCKET_ERRNO);
				error = TRUE;
				break;
			}
			if (wr == 0) {
				lprintf(LOG_WARNING, "%04d <%s> !DATA socket %d disconnected", xfer.ctrl_sock, xfer.user->alias, *xfer.data_sock);
				sockprintf(xfer.ctrl_sock, xfer.ctrl_sess, "426 DATA channel disconnected");
				error = TRUE;
				break;
			}
			lprintf(LOG_ERR, "%04d <%s> !DATA ERROR %d (%d) sending on socket %d"
			        , xfer.ctrl_sock, xfer.user->alias, wr, SOCKET_ERRNO, *xfer.data_sock);
			sockprintf(xfer.ctrl_sock, xfer.ctrl_sess, "451 DATA send error");
			error = TRUE;
			break;
		}
		total += wr;
		*xfer.lastactive = time(NULL);
		//YIELD();
	}

	if ((i = ferror(fp)) != 0)
		lprintf(LOG_ERR, "%04d <%s> !DATA FILE ERROR %d (errno %d %s)"
		        , xfer.ctrl_sock, xfer.user->alias, i, errno, safe_strerror(errno, errstr, sizeof errstr));

	ftp_close_socket(xfer.data_sock, xfer.data_sess, __LINE__);   /* Signal end of file */
	if (startup->options & FTP_OPT_DEBUG_DATA)
		lprintf(LOG_DEBUG, "%04d <%s> DATA socket closed", xfer.ctrl_sock, xfer.user->alias);

	if (!error) {
		dur = (long)(time(NULL) - start);
		cps = (ulong)(dur ? total / dur : total * 2);
		lprintf(LOG_INFO, "%04d <%s> DATA Transfer successful: %" PRIdOFF " bytes sent in %lu seconds (%lu cps)"
		        , xfer.ctrl_sock
		        , xfer.user->alias
		        , total, dur, cps);
		sockprintf(xfer.ctrl_sock, xfer.ctrl_sess, "226 Download complete (%lu cps).", cps);

		if (xfer.dir >= 0 && !xfer.tmpfile) {
			memset(&f, 0, sizeof(f));
			if (!loadfile(&scfg, xfer.dir, getfname(xfer.filename), &f, file_detail_normal)) {
				lprintf(LOG_ERR, "%04d <%s> DATA downloaded: %s (not found in filebase!)"
				        , xfer.ctrl_sock
				        , xfer.user->alias
				        , xfer.filename);
			} else {
				f.hdr.times_downloaded++;
				f.hdr.last_downloaded = time32(NULL);
				updatefile(&scfg, &f);

				lprintf(LOG_INFO, "%04d <%s> DATA downloaded: %s (%u times total)"
				        , xfer.ctrl_sock
				        , xfer.user->alias
				        , xfer.filename
				        , f.hdr.times_downloaded);
				/**************************/
				/* Update Uploader's Info */
				/**************************/
				uploader.number = 0;
				if (f.from_ext != NULL)
					uploader.number = atoi(f.from_ext);
				if (uploader.number == 0)
					uploader.number = matchuser(&scfg, f.from, TRUE /*sysop_alias*/);
				if (uploader.number
				    && uploader.number != xfer.user->number
				    && getuserdat(&scfg, &uploader) == 0
				    && uploader.firston < (time_t)f.hdr.when_imported.time) {
					l = f.cost;
					if (!(scfg.dir[f.dir]->misc & DIR_CDTDL))  /* Don't give credits on d/l */
						l = 0;
					if (scfg.dir[f.dir]->misc & DIR_CDTMIN && cps) { /* Give min instead of cdt */
						mod = ((ulong)(l * (scfg.dir[f.dir]->dn_pct / 100.0)) / cps) / 60;
						adjustuserval(&scfg, uploader.number, USER_MIN, mod);
						sprintf(tmp, "%lu minute", mod);
					} else {
						mod = (ulong)(l * (scfg.dir[f.dir]->dn_pct / 100.0));
						adjustuserval(&scfg, uploader.number, USER_CDT, mod);
						u32toac(mod, tmp, ',');
					}
					if (!(scfg.dir[f.dir]->misc & DIR_QUIET)) {
						const char* prefix = xfer.filepos ? "partially FTP-" : "FTP-";
						addr_len = sizeof(addr);
						if (uploader.level >= SYSOP_LEVEL
						    && getpeername(xfer.ctrl_sock, &addr.addr, &addr_len) == 0
						    && inet_addrtop(&addr, host_ip, sizeof(host_ip)) != NULL)
							SAFEPRINTF2(username, "%s [%s]", xfer.user->alias, host_ip);
						else
							SAFECOPY(username, xfer.user->alias);
						/* Inform uploader of downloaded file */
						if (mod == 0)
							safe_snprintf(str, sizeof(str), text[FreeDownloadUserMsg]
							              , getfname(xfer.filename)
							              , prefix
							              , username);
						else
							safe_snprintf(str, sizeof(str), text[DownloadUserMsg]
							              , getfname(xfer.filename)
							              , prefix
							              , username, tmp);
						putsmsg(&scfg, uploader.number, str);
					}
				}
				mqtt_file_download(&mqtt, xfer.user, f.dir, f.name, total, xfer.client);
				smb_freefilemem(&f);
			}
			if (!xfer.tmpfile && !xfer.delfile && !(scfg.dir[f.dir]->misc & DIR_NOSTAT))
				inc_download_stats(&scfg, 1, (ulong)total);
		}

		if (xfer.credits) {
			user_downloaded(&scfg, xfer.user, 1, total);
			if (xfer.dir >= 0 && !download_is_free(&scfg, xfer.dir, xfer.user, xfer.client))
				subtract_cdt(&scfg, xfer.user, xfer.credits);
		}
	}

	fclose(fp);
	if (ftp_set != NULL && !terminate_server)
		*xfer.inprogress = FALSE;
	if (xfer.tmpfile) {
		if (!(startup->options & FTP_OPT_KEEP_TEMP_FILES))
			ftp_remove(xfer.ctrl_sock, __LINE__, xfer.filename, xfer.user->alias, LOG_ERR);
	}
	else if (xfer.delfile && !error)
		ftp_remove(xfer.ctrl_sock, __LINE__, xfer.filename, xfer.user->alias, LOG_WARNING);

#if defined(SOCKET_DEBUG_SENDTHREAD)
	socket_debug[xfer.ctrl_sock] &= ~SOCKET_DEBUG_SENDTHREAD;
#endif

	thread_down();
}

static void receive_thread(void* arg)
{
	char   str[128];
	char   errstr[256];
	char   buf[8192];
	char   extdesc[LEN_EXTDESC + 1] = "";
	char   tmp[MAX_PATH + 1];
	int    rd;
	off_t  total = 0;
	off_t  last_total = 0;
	ulong  dur;
	ulong  cps;
	BOOL   error = FALSE;
	BOOL   filedat;
	FILE*  fp;
	file_t f;
	xfer_t xfer;
	time_t now;
	time_t start;
	time_t last_report;
	char * estr;

	xfer = *(xfer_t*)arg;
	free(arg);

	SetThreadName("sbbs/ftpReceive");
	thread_up(TRUE /* setuid */);

	if ((fp = fopen(xfer.filename, xfer.append ? "ab" : "wb")) == NULL) {
		lprintf(LOG_ERR, "%04d <%s> !DATA ERROR %d (%s) line %d opening %s"
		        , xfer.ctrl_sock, xfer.user->alias, errno, safe_strerror(errno, errstr, sizeof errstr), __LINE__, xfer.filename);
		sockprintf(xfer.ctrl_sock, xfer.ctrl_sess, "450 ERROR %d (%s) opening %s", errno, safe_strerror(errno, errstr, sizeof errstr), xfer.filename);
		ftp_close_socket(xfer.data_sock, xfer.data_sess, __LINE__);
		*xfer.inprogress = FALSE;
		thread_down();
		return;
	}

	if (xfer.append)
		xfer.filepos = filelength(fileno(fp));

	if (xfer.filepos < 0)
		xfer.filepos = 0;

	*xfer.aborted = FALSE;
	if (xfer.filepos || startup->options & FTP_OPT_DEBUG_DATA)
		lprintf(LOG_DEBUG, "%04d <%s> DATA socket %d receiving %s from offset %" PRIdOFF
		        , xfer.ctrl_sock, xfer.user->alias, *xfer.data_sock, xfer.filename, xfer.filepos);

	fseeko(fp, xfer.filepos, SEEK_SET);

	// Determine the maximum file size to allow, accounting for minimum free space
	char    path[MAX_PATH + 1];
	SAFECOPY(path, xfer.filename);
	*getfname(path) = '\0';
	int64_t avail = getfreediskspace(path, 1);
	if (avail <= scfg.min_dspace)
		avail = 0;
	else
		avail -= scfg.min_dspace;
	int64_t max_fsize = xfer.filepos + avail;
	if (startup->max_fsize > 0 && startup->max_fsize < max_fsize)
		max_fsize = startup->max_fsize;
	if (startup->options & FTP_OPT_DEBUG_DATA)
		lprintf(LOG_DEBUG, "%04d <%s> DATA Limiting uploaded file size to %" PRIu64 " (%s) bytes"
		        , xfer.ctrl_sock, xfer.user->alias, max_fsize
		        , byte_estimate_to_str(max_fsize, tmp, sizeof(tmp), 1, 1));

	last_report = start = time(NULL);
	while (1) {

		now = time(NULL);

		/* Periodic progress report */
		if (total && now >= last_report + XFER_REPORT_INTERVAL) {
			if (xfer.filepos)
				sprintf(str, " from offset %" PRIdOFF, xfer.filepos);
			else
				str[0] = 0;
			lprintf(LOG_INFO, "%04d <%s> DATA Received %" PRIdOFF " bytes of %s (%lu cps)%s"
			        , xfer.ctrl_sock
			        , xfer.user->alias
			        , total, xfer.filename
			        , (ulong)((total - last_total) / (now - last_report))
			        , str);
			last_total = total;
			last_report = now;
		}
		if (xfer.filepos + total > max_fsize) {
			lprintf(LOG_WARNING, "%04d <%s> !DATA received %" PRIdOFF " bytes of %s exceeds maximum allowed (%" PRIu64 " bytes)"
			        , xfer.ctrl_sock, xfer.user->alias, xfer.filepos + total, xfer.filename, startup->max_fsize);
			sockprintf(xfer.ctrl_sock, xfer.ctrl_sess, "552 File size exceeds maximum allowed (%" PRIu64 " bytes)", startup->max_fsize);
			error = TRUE;
			break;
		}
		if (*xfer.aborted == TRUE) {
			lprintf(LOG_WARNING, "%04d <%s> !DATA Transfer aborted", xfer.ctrl_sock, xfer.user->alias);
			/* Send NAK */
			sockprintf(xfer.ctrl_sock, xfer.ctrl_sess, "426 Transfer aborted.");
			error = TRUE;
			break;
		}
		if (ftp_set == NULL || terminate_server) {
			lprintf(LOG_WARNING, "%04d <%s> !DATA Transfer locally aborted", xfer.ctrl_sock, xfer.user->alias);
			/* Send NAK */
			sockprintf(xfer.ctrl_sock, xfer.ctrl_sess, "426 Transfer locally aborted.");
			error = TRUE;
			break;
		}

		/* Check socket for readability */
		if (!socket_readable(*xfer.data_sock, 1000))
			continue;

#if defined(SOCKET_DEBUG_RECV_BUF)
		socket_debug[xfer.ctrl_sock] |= SOCKET_DEBUG_RECV_BUF;
#endif
		if (*xfer.data_sess != -1) {
			int status = cryptPopData(*xfer.data_sess, buf, sizeof(buf), &rd);
			if (status != CRYPT_OK) {
				GCES(status, *xfer.data_sock, *xfer.data_sess, estr, "popping data");
				if (status != CRYPT_ERROR_COMPLETE)
					rd = SOCKET_ERROR;
			}
		}
		else {
			rd = recv(*xfer.data_sock, buf, sizeof(buf), 0);