path: root/ssh/tomsg_clientlib.c

#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <ctype.h>
#include <inttypes.h>
#include <unistd.h>
#include <assert.h>
#include <errno.h>
#include <pthread.h>
#include <poll.h>
#include "tomsg_clientlib.h"
#include "sshnc.h"
#include "string_view.h"


struct inflight {
	int64_t tag;

	// Partially filled in, depending on the event type
	struct tomsg_event event;
};

struct tomsg_client {
	struct sshnc_client *conn;

	// Receive buffer for the ssh connection
	size_t buffer_len, buffer_cap;
	char *buffer;
	size_t buffer_newline_cursor;  // no newlines before this index

	int64_t next_tag;

	size_t inflight_num, inflight_cap;
	struct inflight *inflight;
};

enum tomsg_async_connect_state {
	STATE_CONNECTING,
	STATE_KEY_RECEIVED,
	STATE_ACCEPTED,
};

// Only field written to by the thread is 'client'.
struct tomsg_async_connect {
	enum tomsg_async_connect_state state;

	char *hostname;
	int port;

	// Two pipes for communicating with the thread
	int host_w, host_r, thread_w, thread_r;

	struct tomsg_client *client;  // filled by thread in case all is successful
};

static size_t min_size_t(size_t a, size_t b) { return a < b ? a : b; }

static bool hasspacelf(const char *string) {
	for (size_t i = 0; string[i]; i++)
		if (string[i] == ' ' || string[i] == '\n') return true;
	return false;
}

static bool haslf(const char *string) {
	for (size_t i = 0; string[i]; i++)
		if (string[i] == '\n') return true;
	return false;
}

__attribute__((warn_unused_result))
static enum tomsg_retval add_inflight(
		struct tomsg_client *client, int64_t tag, struct tomsg_event event) {
	if (client->inflight_num == client->inflight_cap) {
		client->inflight_cap *= 2;
		struct inflight *new_inflight = realloc(
				client->inflight, client->inflight_cap * sizeof(struct inflight));
		if (!new_inflight) {
			// Just don't send the message; hope the application can handle that
			tomsg_event_nullify(event);
			return TOMSG_ERR_MEMORY;
		}
		client->inflight = new_inflight;
	}

	client->inflight[client->inflight_num++] = (struct inflight){
		.tag = tag,
		.event = event,
	};

	return TOMSG_OK;
}

static enum tomsg_retval receive_more_data(struct tomsg_client *client) {
	if (client->buffer_cap - client->buffer_len < 256) {
		client->buffer_cap *= 2;
		char *new_buffer = realloc(client->buffer, client->buffer_cap);
		if (!new_buffer) {
			// Cannot safely continue; we also can't "just" clear the buffer
			// and start parsing again, since we may detect commands in the
			// middle of messages, which, even if not an RCE vector, would not
			// be good. Thus, throw everything away and close the connection
			// now, before stuff goes wrong.
			free(client->buffer);
			client->buffer = NULL;
			sshnc_close(client->conn);
			client->conn = NULL;
			return TOMSG_ERR_MEMORY;
		}
		client->buffer = new_buffer;
	}

	size_t recvlen = 0;
	const enum sshnc_retval ret = sshnc_maybe_recv(
		client->conn,
		client->buffer_cap - client->buffer_len,
		client->buffer + client->buffer_len,
		&recvlen
	);

	if (ret == SSHNC_EOF) return TOMSG_ERR_CLOSED;
	if (ret == SSHNC_AGAIN) return TOMSG_ERR_AGAIN;
	if (ret != SSHNC_OK) return TOMSG_ERR_TRANSPORT;

	client->buffer_len += recvlen;

	return TOMSG_OK;
}

// Returns the first line in the buffer, if any. Updates buffer_newline_cursor
// to point at the newline after that line, or at the end of the buffer
// otherwise.
static struct string_view extract_next_line(struct tomsg_client *client) {
	for (size_t i = client->buffer_newline_cursor; i < client->buffer_len; i++) {
		if (client->buffer[i] == '\n') {
			struct string_view sv = string_view(client->buffer, i);

			client->buffer_newline_cursor = i;
			return sv;
		}
	}

	client->buffer_newline_cursor = client->buffer_len;
	return string_view(NULL, 0);
}

// Assumes buffer_newline_cursor points at a newline; removes the part ending
// at that newline from the buffer.
static void splice_scanned_buffer_part(struct tomsg_client *client) {
	const size_t start = min_size_t(client->buffer_newline_cursor + 1, client->buffer_len);
	const size_t restlen = client->buffer_len - start;
	memmove(client->buffer, client->buffer + start, restlen);
	client->buffer_len = restlen;
	client->buffer_newline_cursor = 0;
}

const char* tomsg_print_hash(const unsigned char *hash, size_t length) {
	return sshnc_print_hash(hash, length);
}

const char* tomsg_strerror(enum tomsg_retval code) {
	switch (code) {
		case TOMSG_OK: return "Success";
		case TOMSG_ERR_CONNECT: return "Server refused connection";
		case TOMSG_ERR_UNTRUSTED: return "Hostkey was rejected";
		case TOMSG_ERR_CLOSED: return "Server connection unexpectedly closed";
		case TOMSG_ERR_VERSION: return "Server protocol version incompatible";
		case TOMSG_ERR_TRANSPORT: return "Error in the underlying SSH transport";
		case TOMSG_ERR_SPACE: return "Argument contained space or LF, while it could not";
		case TOMSG_ERR_AGAIN: return "No events for now, poll(2) and try again";
		case TOMSG_ERR_PARSE: return "Could not parse line from server, line ignored";
		case TOMSG_ERR_MEMORY: return "Error allocating memory";
	}

	return "?unknown error?";
}

static enum tomsg_retval version_negotiation(struct tomsg_client *client) {
	if (!client->conn) return TOMSG_ERR_CLOSED;

	const enum sshnc_retval retssh = sshnc_send(client->conn, "ver version 3\n", 14);
	if (retssh == SSHNC_EOF) return TOMSG_ERR_CLOSED;
	if (retssh != SSHNC_OK) return TOMSG_ERR_TRANSPORT;

	struct pollfd pfd;
	pfd.fd = sshnc_poll_fd(client->conn);
	pfd.events = POLLIN;

	while (true) {
		const int retpoll = poll(&pfd, 1, -1);
		if (retpoll < 0) return TOMSG_ERR_TRANSPORT;
		if (retpoll == 0) continue;

		const enum tomsg_retval retmsg = receive_more_data(client);
		if (retmsg == TOMSG_ERR_AGAIN) continue;
		if (retmsg != TOMSG_OK) return retmsg;

		const struct string_view line = extract_next_line(client);
		if (line.s != NULL) {
			if (sv_equals(line, "ver ok")) {
				splice_scanned_buffer_part(client);
				return TOMSG_OK;
			} else {
				sshnc_close(client->conn);
				client->conn = NULL;
				return TOMSG_ERR_VERSION;
			}
		}
	}
}

enum tomsg_retval tomsg_connect(
		const char *hostname, int port,
		tomsg_hostkey_checker_t checker, void *userdata,
		struct tomsg_client **clientp) {
	// In case we throw an error along the way
	*clientp = NULL;

	struct sshnc_client *conn;
	enum sshnc_retval ret = sshnc_connect(
			hostname, port, "tomsg", "tomsg", checker, userdata, &conn);

	if (ret == SSHNC_ERR_CONNECT) return TOMSG_ERR_CONNECT;
	if (ret == SSHNC_ERR_UNTRUSTED) return TOMSG_ERR_UNTRUSTED;
	if (ret != SSHNC_OK) return TOMSG_ERR_TRANSPORT;

	struct tomsg_client *client = malloc(sizeof(struct tomsg_client));
	if (!client) return TOMSG_ERR_MEMORY;
	client->conn = conn;
	client->buffer_len = 0;
	client->buffer_cap = 1024;
	client->buffer = malloc(client->buffer_cap);
	if (!client->buffer) { free(client); return TOMSG_ERR_MEMORY; }
	client->buffer_newline_cursor = 0;
	client->next_tag = 1;
	client->inflight_num = 0;
	client->inflight_cap = 2;
	client->inflight = malloc(client->inflight_cap * sizeof(struct inflight));
	if (!client->inflight) { free(client->buffer); free(client); return TOMSG_ERR_MEMORY; }

	*clientp = client;

	return version_negotiation(client);
}

// Returns whether successful
static bool writeall(int fd, const unsigned char *buffer, size_t length) {
	size_t cursor = 0;
	while (cursor < length) {
		ssize_t nw = write(fd, buffer + cursor, length - cursor);
		if (nw < 0) {
			if (errno == EINTR) continue;
			return false;
		}
		if (nw == 0) return false;
		cursor += nw;
	}
	return true;
}

// Returns whether successful
static bool readall(int fd, unsigned char *buffer, size_t length) {
	size_t cursor = 0;
	while (cursor < length) {
		ssize_t nr = read(fd, buffer + cursor, length - cursor);
		if (nr < 0) {
			if (errno == EINTR) continue;
			return false;
		}
		if (nr == 0) return false;
		cursor += nr;
	}
	return true;
}

// Async socket protocol:
// Sizes are always indicated using a host-order 8-byte signed integer.
// - Thread sends an error byte; if that's TOMSG_OK, it is followed by size and
//   hashbytes of the hostkey. Otherwise, the thread exits.
// - Host sends one byte: 0 for reject, 1 for accept.
// - For reject, the thread closes the connection and exits. For accept, the
//   thread negotiates the protocol version and if successful, initialises and
//   populates a client structure in the state, and sends a TOMSG_OK byte. If
//   unsuccessful, sends an error byte and exits.
// After the thread sends either an error byte or the final OK byte, or after
// it has received a reject message, it will not access 'state' anymore; it can
// thus be freed by the host.

static bool async_hostkey_checker(const unsigned char *hash, size_t length, void *state_) {
	struct tomsg_async_connect *state = state_;

	unsigned char buffer[256];
	if (9 + length > sizeof buffer) { assert(false); return false; }
	buffer[0] = TOMSG_OK;
	*(int64_t*)&buffer[1] = length;
	memcpy(buffer + 9, hash, length);

	if (!writeall(state->thread_w, buffer, 9 + length)) return false;

	char response;
	ssize_t nr = -1;
	while (nr < 0) {
		nr = read(state->thread_r, &response, 1);
		if (nr == 0 || (nr < 0 && errno != EINTR)) break;
	}
	if (nr < 0) return false;

	return response == 1;
}

static void* async_connect_thread_entry(void *state_) {
	struct tomsg_async_connect *state = state_;

	const int thread_r = state->thread_r;
	const int thread_w = state->thread_w;

	struct sshnc_client *conn;
	enum sshnc_retval ret = sshnc_connect(
			state->hostname, state->port, "tomsg", "tomsg",
			async_hostkey_checker, state, &conn);

	enum tomsg_retval sendret;
	if (ret == SSHNC_ERR_CONNECT) sendret = TOMSG_ERR_CONNECT;
	else if (ret == SSHNC_ERR_UNTRUSTED) sendret = TOMSG_ERR_UNTRUSTED;
	else if (ret != SSHNC_OK) sendret = TOMSG_ERR_TRANSPORT;
	else sendret = TOMSG_OK;

	// If sendret is TOMSG_ERR_UNTRUSTED, the host may free 'state' at this point. Thus, don't access it, please.

	struct tomsg_client *client = NULL;
	unsigned char byte;

	if (sendret == TOMSG_ERR_UNTRUSTED) goto pipe_return;
	if (sendret != TOMSG_OK) goto send_error_return;

	client = calloc(1, sizeof(struct tomsg_client));
	if (!client) { sendret = TOMSG_ERR_MEMORY; goto send_error_return; }
	client->conn = conn;
	client->buffer_len = 0;
	client->buffer_cap = 1024;
	client->buffer = malloc(client->buffer_cap);
	if (!client->buffer) { sendret = TOMSG_ERR_MEMORY; goto send_error_return; }
	client->buffer_newline_cursor = 0;
	client->next_tag = 1;
	client->inflight_num = 0;
	client->inflight_cap = 2;
	client->inflight = malloc(client->inflight_cap * sizeof(struct inflight));
	if (!client->inflight) { sendret = TOMSG_ERR_MEMORY; goto send_error_return; }

	enum tomsg_retval versionret = version_negotiation(client);
	if (versionret != TOMSG_OK) { sendret = versionret; goto send_error_return; }

	state->client = client;

	byte = TOMSG_OK;
	// After the writeall, it is forbidden to access anything in 'state',
	// because it may have been freed by the host at this point.
	if (!writeall(thread_w, &byte, 1)) goto free_client;
	goto pipe_return;

send_error_return:
	// After this, it is forbidden to access anything in 'state', because it
	// may have been freed by the host at this point.
	byte = sendret;
	writeall(thread_w, &byte, 1);
free_client:
	if (conn) sshnc_close(conn);
	if (client) {
		if (client->buffer) free(client->buffer);
		if (client->inflight) free(client->inflight);
		free(client);
	}
pipe_return:
	close(thread_r);
	close(thread_w);
	return NULL;
}

enum tomsg_retval tomsg_async_connect(
		const char *hostname, int port,
		struct tomsg_async_connect **clientp) {
	// In case we throw an error along the way
	*clientp = NULL;

	struct tomsg_async_connect *client = malloc(sizeof(struct tomsg_async_connect));
	if (!client) return TOMSG_ERR_MEMORY;

	client->state = STATE_CONNECTING;

	client->hostname = strdup(hostname);
	client->port = port;

	pthread_attr_t attr;
	if (pthread_attr_init(&attr) < 0) {
		free(client->hostname);
		free(client);
		if (errno == ENOMEM) return TOMSG_ERR_MEMORY;
		return TOMSG_ERR_CONNECT;
	}

	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

	int pipeHT[2] = {-1, -1}, pipeTH[2];
	if (pipe(pipeHT) < 0 || pipe(pipeTH) < 0) {
		if (pipeHT[0] != -1) { close(pipeHT[0]); close(pipeHT[1]); }
		free(client->hostname);
		free(client);
		return TOMSG_ERR_CONNECT;
	}

	client->host_w = pipeHT[1];
	client->thread_r = pipeHT[0];
	client->thread_w = pipeTH[1];
	client->host_r = pipeTH[0];

	pthread_t thread;
	if (pthread_create(&thread, &attr, async_connect_thread_entry, client) < 0) {
		pthread_attr_destroy(&attr);
		close(client->host_w); close(client->host_r);
		close(client->thread_w); close(client->thread_r);
		free(client->hostname);
		free(client);
		return TOMSG_ERR_CONNECT;
	}

	*clientp = client;
	return TOMSG_OK;
}

static bool check_readable(int fd) {
	struct pollfd pfd;
	pfd.fd = fd;
	pfd.events = POLLIN;
	pfd.revents = 0;
	poll(&pfd, 1, 0);
	return pfd.revents & (POLLIN | POLLHUP | POLLERR);
}

void tomsg_async_connect_event_nullify(struct tomsg_async_connect_event event) {
	switch (event.type) {
		case TOMSG_AC_HOSTKEY:
			free(event.key.hostkey);
			break;

		case TOMSG_AC_SUCCESS:
			tomsg_close(event.client);
			break;
	}
}

enum tomsg_retval tomsg_async_connect_next_event(
	struct tomsg_async_connect *client,
	struct tomsg_async_connect_event *event  // output
) {
	enum tomsg_retval final_retval;

	switch (client->state) {
		case STATE_CONNECTING: {
			if (!check_readable(client->host_r)) return TOMSG_ERR_AGAIN;

			unsigned char byte;
			if (!readall(client->host_r, &byte, 1)) goto non_recoverable_error;
			if (byte != TOMSG_OK) {
				final_retval = byte;
				goto free_return;
			}

			// Now read the hostkey from the pipe
			int64_t length;
			if (!readall(client->host_r, (unsigned char*)&length, 8)) goto non_recoverable_error;
			unsigned char *hash = malloc(length);
			if (!hash) goto non_recoverable_error;
			if (!readall(client->host_r, hash, length)) goto non_recoverable_error;

			client->state = STATE_KEY_RECEIVED;
			event->type = TOMSG_AC_HOSTKEY;
			event->key.hostkey = hash;
			event->key.length = length;
			return TOMSG_OK;
		}

		case STATE_KEY_RECEIVED:
			return TOMSG_ERR_AGAIN;  // you need to accept or reject!

		case STATE_ACCEPTED: {
			if (!check_readable(client->host_r)) return TOMSG_ERR_AGAIN;

			unsigned char byte;
			if (!readall(client->host_r, &byte, 1)) goto non_recoverable_error;
			if (byte == TOMSG_OK) {
				event->type = TOMSG_AC_SUCCESS;
				event->client = client->client;
				client->client = NULL;
				final_retval = TOMSG_OK;
				goto free_return;
			} else {
				final_retval = byte;
				goto free_return;
			}
			break;
		}
	}

non_recoverable_error:
	// Don't even know how to handle this correctly; let's just hope the thread kills itself somehow
	close(client->host_w);
	close(client->host_r);
	return TOMSG_ERR_TRANSPORT;

free_return:
	free(client->hostname);
	close(client->host_w);
	close(client->host_r);
	if (client->client) tomsg_close(client->client);
	return final_retval;
}

enum tomsg_retval tomsg_async_connect_accept(struct tomsg_async_connect *client, bool accept) {
	enum tomsg_retval final_retval;
	if (client->state != STATE_KEY_RECEIVED) {
		final_retval = TOMSG_ERR_TRANSPORT;  // shrug
		goto free_return;
	}

	unsigned char byte = accept ? 1 : 0;
	if (!writeall(client->host_w, &byte, 1)) {
		goto non_recoverable_error;
	}

	if (!accept) {
		final_retval = TOMSG_ERR_UNTRUSTED;
		goto free_return;
	}

	client->state = STATE_ACCEPTED;

	return TOMSG_OK;

non_recoverable_error:
	// Don't even know how to handle this correctly; let's just hope the thread kills itself somehow
	close(client->host_w);
	close(client->host_r);
	return TOMSG_ERR_TRANSPORT;

free_return:
	free(client->hostname);
	close(client->host_w);
	close(client->host_r);
	if (client->client) tomsg_close(client->client);
	return final_retval;
}

int tomsg_async_connect_poll_fd(const struct tomsg_async_connect *client) {
	return client->host_r;
}

void tomsg_close(struct tomsg_client *client) {
	if (client->conn) sshnc_close(client->conn);
	free(client->buffer);
	if (client->inflight) {
		for (size_t i = 0; i < client->inflight_num; i++) {
			tomsg_event_nullify(client->inflight[i].event);
		}
	}
	free(client->inflight);
	free(client);
}

int tomsg_poll_fd(const struct tomsg_client *client) {
	if (!client->conn) return -1;
	return sshnc_poll_fd(client->conn);
}

static void history_message_nullify(struct history_message message) {
	free(message.username); message.username = NULL;
	free(message.message); message.message = NULL;
}

void tomsg_event_nullify(struct tomsg_event event) {
	if (event.error) free(event.error);
	event.error = NULL;

	switch (event.type) {
		case TOMSG_EV_LOGOUT:
		case TOMSG_EV_PING:
		case TOMSG_EV_USER_ACTIVE:
		case TOMSG_EV_SEND:
			break;

		case TOMSG_EV_REGISTER:
		case TOMSG_EV_LOGIN:
			free(event.login.username);
			break;

		case TOMSG_EV_INVITE:
		case TOMSG_EV_PUSH_JOIN:
			free(event.join.room_name);
			free(event.join.username);
			break;

		case TOMSG_EV_IS_ONLINE:
		case TOMSG_EV_PUSH_ONLINE:
			free(event.is_online.username);
			break;

		case TOMSG_EV_CREATE_ROOM:
			free(event.create_room.room_name);
			break;

		case TOMSG_EV_LEAVE_ROOM:
			free(event.leave_room.room_name);
			break;

		case TOMSG_EV_LIST_ROOMS:
			if (event.list_rooms.rooms != NULL) {
				for (int64_t i = 0; i < event.list_rooms.count; i++) {
					free(event.list_rooms.rooms[i]);
				}
			}
			free(event.list_rooms.rooms);
			break;

		case TOMSG_EV_LIST_MEMBERS:
			if (event.list_members.members != NULL) {
				for (int64_t i = 0; i < event.list_members.count; i++) {
					free(event.list_members.members[i]);
				}
			}
			free(event.list_members.members);
			break;

		case TOMSG_EV_HISTORY:
			free(event.history.room_name);
			if (event.history.messages != NULL) {
				for (int64_t i = 0; i < event.history.count; i++) {
					history_message_nullify(event.history.messages[i]);
				}
			}
			free(event.history.messages);
			break;

		case TOMSG_EV_GET_MESSAGE:
			free(event.get_message.room_name);
			history_message_nullify(event.get_message.message);
			break;

		case TOMSG_EV_PUSH_MESSAGE:
			free(event.push_message.room_name);
			history_message_nullify(event.push_message.message);
			break;

		case TOMSG_EV_PUSH_INVITE:
			free(event.push_invite.room_name);
			free(event.push_invite.inviter);
			break;

		case TOMSG_EV_PUSH_LEAVE:
			free(event.push_leave.room_name);
			free(event.push_leave.username);
			break;
	}
}

static bool find_extract_inflight(struct tomsg_client *client, int64_t tag, struct inflight *output) {
	for (size_t i = 0; i < client->inflight_num; i++) {
		if (client->inflight[i].tag == tag) {
			*output = client->inflight[i];
			if (i < client->inflight_num - 1) {
				client->inflight[i] = client->inflight[client->inflight_num - 1];
			}
			client->inflight_num--;
			return true;
		}
	}

	return false;
}

// Returns TOMSG_ERR_AGAIN in case the line generated no output event.
static enum tomsg_retval handle_line(
	struct tomsg_client *client,
	struct string_view line,
	struct tomsg_event *eventp
) {
	const struct string_view tagstr = sv_tokenise_word(&line);
	const struct string_view command = sv_tokenise_word(&line);

	// These parse macro's directly write to eventp.
#define PARSE_I64(memb_) \
		do { if (!sv_parse_i64(sv_tokenise_word(&line), &eventp -> memb_)) return TOMSG_ERR_PARSE; } while (0)
#define PARSE_WORD(memb_) \
		do { if (!sv_copy(sv_tokenise_word(&line), &eventp -> memb_)) return TOMSG_ERR_PARSE; } while (0)
#define PARSE_RESTSTRING(memb_) \
		do { if (!sv_copy(line, &eventp -> memb_)) return TOMSG_ERR_PARSE; } while (0)
#define EXPECT_FINISH() \
		do { if (!sv_is_empty(line)) return TOMSG_ERR_PARSE; } while (0)

	if (sv_equals(tagstr, "_push")) {
		// Ordered by rought expected occurrence probability
		if (sv_equals(command, "ping")) {
			return TOMSG_ERR_AGAIN;
		} else if (sv_equals(command, "message")) {
			eventp->type = TOMSG_EV_PUSH_MESSAGE;
			PARSE_WORD(push_message.room_name);
			PARSE_WORD(push_message.message.username);
			PARSE_I64(push_message.message.timestamp);
			PARSE_I64(push_message.message.msgid);
			PARSE_I64(push_message.message.replyid);
			PARSE_RESTSTRING(push_message.message.message);
			return TOMSG_OK;
		} else if (sv_equals(command, "online")) {
			eventp->type = TOMSG_EV_PUSH_ONLINE;
			PARSE_I64(is_online.online_count);
			PARSE_WORD(is_online.username);
			EXPECT_FINISH();
			return TOMSG_OK;
		} else if (sv_equals(command, "invite")) {
			eventp->type = TOMSG_EV_PUSH_INVITE;
			PARSE_WORD(push_invite.room_name);
			PARSE_WORD(push_invite.inviter);
			EXPECT_FINISH();
			return TOMSG_OK;
		} else if (sv_equals(command, "join")) {
			eventp->type = TOMSG_EV_PUSH_JOIN;
			PARSE_WORD(join.room_name);
			PARSE_WORD(join.username);
			EXPECT_FINISH();
			return TOMSG_OK;
		} else if (sv_equals(command, "leave")) {
			eventp->type = TOMSG_EV_PUSH_LEAVE;
			PARSE_WORD(push_leave.room_name);
			PARSE_WORD(push_leave.username);
			EXPECT_FINISH();
			return TOMSG_OK;
		} else {
			// Unknown command, let's just ignore
			return TOMSG_ERR_AGAIN;
		}
	}

	// These parse functions are now not valid anymore, since we can't directly
	// write to eventp anymore, and additionally need to manage the event
	// structure we will have around in 'inflight'.
#undef PARSE_I64
#undef PARSE_WORD
#undef PARSE_RESTSTRING
#undef EXPECT_FINISH

	int64_t tag;
	struct inflight inflight;
	if (!sv_parse_i64(tagstr, &tag) || !find_extract_inflight(client, tag, &inflight)) {
		// Unknown tag, let's just ignore
		return TOMSG_ERR_AGAIN;
	}

	// These macros correctly dispose of, or persist, the event structure before returning.
#define CLEANUP_RETURN(ret_) do { tomsg_event_nullify(inflight.event); return (ret_); } while (0)
#define SUCCESS_RETURN() do { *eventp = inflight.event; return TOMSG_OK; } while (0)

	if (sv_equals(command, "error")) {
		if (!sv_copy(line, &inflight.event.error)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
		SUCCESS_RETURN();
	}

	// Since the response is not 'error', at least we're not in an error case.
	// Thus, we can assert the other, successful response type in each of the
	// event arms below.
	inflight.event.error = NULL;

	switch (inflight.event.type) {
		case TOMSG_EV_REGISTER:
		case TOMSG_EV_LOGIN:
		case TOMSG_EV_LOGOUT:
		case TOMSG_EV_INVITE:
		case TOMSG_EV_LEAVE_ROOM:
			if (!sv_equals(command, "ok")) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			SUCCESS_RETURN();

		case TOMSG_EV_LIST_ROOMS:
		case TOMSG_EV_LIST_MEMBERS: {
			if (!sv_equals(command, "list")) CLEANUP_RETURN(TOMSG_ERR_PARSE);

			int64_t count;
			char **words;
			if (!sv_parse_i64(sv_tokenise_word(&line), &count)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (count < 0 || count > (1 << 30)) CLEANUP_RETURN(TOMSG_ERR_PARSE);

			words = calloc(count, sizeof(char*));
			if (!words) CLEANUP_RETURN(TOMSG_ERR_MEMORY);
			for (int64_t i = 0; i < count; i++) {
				if (!sv_copy(sv_tokenise_word(&line), &words[i])) {
					for (int64_t j = 0; j < i; j++) free(words[j]);
					CLEANUP_RETURN(TOMSG_ERR_PARSE);
				}
			}

			// There might be words after the last one in the list, but we'll
			// just ignore them for simplicity's sake.

			if (inflight.event.type == TOMSG_EV_LIST_ROOMS) {
				inflight.event.list_rooms.count = count;
				inflight.event.list_rooms.rooms = words;
			} else {
				inflight.event.list_members.count = count;
				inflight.event.list_members.members = words;
			}

			SUCCESS_RETURN();
		}

		case TOMSG_EV_CREATE_ROOM:
			if (!sv_equals(command, "name")) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_copy(sv_tokenise_word(&line), &inflight.event.create_room.room_name)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_is_empty(line)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			SUCCESS_RETURN();

		case TOMSG_EV_SEND:
			if (!sv_equals(command, "number")) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_parse_i64(sv_tokenise_word(&line), &inflight.event.send.msgid)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_is_empty(line)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			SUCCESS_RETURN();

		case TOMSG_EV_HISTORY:
			if (sv_equals(command, "history")) {
				int64_t count;
				if (!sv_parse_i64(sv_tokenise_word(&line), &count)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
				if (count < 0 || count > (1 << 30)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
				inflight.event.history.count = count;

				// Use calloc() to ensure the embedded pointers are NULL, in case we free it early
				inflight.event.history.messages = calloc(count, sizeof(struct history_message));
				if (!inflight.event.history.messages) CLEANUP_RETURN(TOMSG_ERR_MEMORY);

				// If there are no history_message's coming, complete early
				if (count == 0) {
					SUCCESS_RETURN();
				}

				// Re-add the tag for the history_message events
				enum tomsg_retval ret = add_inflight(client, tag, inflight.event);
				if (ret != TOMSG_OK) return ret;
				return TOMSG_ERR_AGAIN;  // don't cleanup, we re-submitted the event!
			} else if (sv_equals(command, "history_message")) {
				int64_t index;
				if (!sv_parse_i64(sv_tokenise_word(&line), &index)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
				if (index < 0 || index >= inflight.event.history.count) CLEANUP_RETURN(TOMSG_ERR_PARSE);

				// Note that early-exiting in case of parse errors is valid here
				// (and everywhere): the messages buffer has been allocated with
				// calloc(), so tomsg_event_nullify() will cleanup up precisely
				// everything that we haven't filled in yet.
				struct history_message *msg = &inflight.event.history.messages[index];
				if (!sv_equals(sv_tokenise_word(&line), inflight.event.history.room_name)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
				if (!sv_copy(sv_tokenise_word(&line), &msg->username)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
				if (!sv_parse_i64(sv_tokenise_word(&line), &msg->timestamp)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
				if (!sv_parse_i64(sv_tokenise_word(&line), &msg->msgid)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
				if (!sv_parse_i64(sv_tokenise_word(&line), &msg->replyid)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
				if (!sv_copy(line, &msg->message)) CLEANUP_RETURN(TOMSG_ERR_PARSE);

				// If there are more history_message events coming, re-add the tag for them
				if (index == inflight.event.history.count - 1) {
					SUCCESS_RETURN();  // done with the whole history listing
				} else {
					enum tomsg_retval ret = add_inflight(client, tag, inflight.event);
					if (ret != TOMSG_OK) return ret;
					return TOMSG_ERR_AGAIN;  // don't cleanup, we re-submitted the event!
				}
			} else {
				CLEANUP_RETURN(TOMSG_ERR_PARSE);
			}

		case TOMSG_EV_GET_MESSAGE:
			if (!sv_equals(command, "message")) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_copy(sv_tokenise_word(&line), &inflight.event.get_message.room_name)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_copy(sv_tokenise_word(&line), &inflight.event.get_message.message.username)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_parse_i64(sv_tokenise_word(&line), &inflight.event.get_message.message.timestamp)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_parse_i64(sv_tokenise_word(&line), &inflight.event.get_message.message.msgid)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_parse_i64(sv_tokenise_word(&line), &inflight.event.get_message.message.replyid)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_copy(line, &inflight.event.get_message.message.message)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			SUCCESS_RETURN();

		case TOMSG_EV_PING:
			if (!sv_equals(command, "pong")) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_is_empty(line)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			SUCCESS_RETURN();

		case TOMSG_EV_IS_ONLINE:
			if (!sv_equals(command, "number")) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_parse_i64(sv_tokenise_word(&line), &inflight.event.is_online.online_count)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_is_empty(line)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			SUCCESS_RETURN();

		case TOMSG_EV_USER_ACTIVE:
			if (!sv_equals(command, "ok")) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			if (!sv_is_empty(line)) CLEANUP_RETURN(TOMSG_ERR_PARSE);
			SUCCESS_RETURN();

		case TOMSG_EV_PUSH_ONLINE:
		case TOMSG_EV_PUSH_MESSAGE:
		case TOMSG_EV_PUSH_INVITE:
		case TOMSG_EV_PUSH_JOIN:
		case TOMSG_EV_PUSH_LEAVE:
			// What? Why did we put a push event in the inflight list? That's a bug.
			assert(false);
	}

#undef CLEANUP_RETURN
#undef SUCCESS_RETURN

	// Should not be reachable; either the switch above missed a 'return'
	// statement somewhere, which is a bug, or the event type was not in the
	// tomsg_event_type enum, which is also a bug.
	assert(false);
}

// If the buffer contains a full line, handles that line, stores the received
// event in eventp, and removes the line from the buffer. Otherwise, returns
// TOMSG_ERR_AGAIN.
static enum tomsg_retval find_handle_next_line(
		struct tomsg_client *client, struct tomsg_event *eventp) {

	// Loop while a line was handled without producing an output event
	while (true) {
		struct string_view line = extract_next_line(client);
		if (line.s == NULL) return TOMSG_ERR_AGAIN;  // out of data

		const enum tomsg_retval ret = handle_line(client, line, eventp);
		splice_scanned_buffer_part(client);
		if (ret == TOMSG_ERR_AGAIN) continue;  // handled but no output
		return ret;  // handled with output, or error thrown
	}
}

enum tomsg_retval tomsg_next_event(struct tomsg_client *client, struct tomsg_event *eventp) {
	if (!client->conn) return TOMSG_ERR_CLOSED;

	// If we already have a line waiting in the buffer, handle it now.
	enum tomsg_retval ret = find_handle_next_line(client, eventp);
	if (ret != TOMSG_ERR_AGAIN) return ret;

	// Otherwise, try to receive more data until either there is no more data
	// anymore, or we have a complete event.
	while (true) {
		ret = receive_more_data(client);
		if (ret != TOMSG_OK) return ret;

		// Handle any line that might be completed at this point. If that
		// produces an event or an error, return it; otherwise loop and receive
		// more data.
		ret = find_handle_next_line(client, eventp);
		if (ret != TOMSG_ERR_AGAIN) return ret;
	}
}

#define SEND_FMT0(client, tag, fmt) do { \
		char buffer_[128]; \
		int length_ = snprintf(buffer_, sizeof buffer_, "%" PRIi64 " " fmt "\n", (tag)); \
		assert(length_ < (int)sizeof buffer_); \
		enum sshnc_retval ret_ = sshnc_send((client)->conn, buffer_, length_); \
		if (ret_ == SSHNC_EOF) return TOMSG_ERR_CLOSED; \
		if (ret_ != SSHNC_OK) return TOMSG_ERR_TRANSPORT; \
	} while (0)

#define SEND_FMT(client, tag, fmt, ...) do { \
		char *buffer_; \
		int length_ = asprintf(&buffer_, "%" PRIi64 " " fmt "\n", (tag), __VA_ARGS__); \
		if (length_ < 0) return TOMSG_ERR_MEMORY; \
		enum sshnc_retval ret_ = sshnc_send((client)->conn, buffer_, length_); \
		free(buffer_); \
		if (ret_ == SSHNC_EOF) return TOMSG_ERR_CLOSED; \
		if (ret_ != SSHNC_OK) return TOMSG_ERR_TRANSPORT; \
	} while (0)

enum tomsg_retval tomsg_register(
		struct tomsg_client *client, const char *username, const char *password) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	if (hasspacelf(username)) return TOMSG_ERR_SPACE;
	if (haslf(password)) return TOMSG_ERR_SPACE;
	const int64_t tag = client->next_tag++;
	SEND_FMT(client, tag, "register %s %s", username, password);
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_REGISTER,
		.login.username = strdup(username),
	};
	return add_inflight(client, tag, event);
}

enum tomsg_retval tomsg_login(
		struct tomsg_client *client, const char *username, const char *password) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	if (hasspacelf(username)) return TOMSG_ERR_SPACE;
	if (haslf(password)) return TOMSG_ERR_SPACE;
	const int64_t tag = client->next_tag++;
	SEND_FMT(client, tag, "login %s %s", username, password);
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_LOGIN,
		.login.username = strdup(username),
	};
	return add_inflight(client, tag, event);
}

enum tomsg_retval tomsg_logout(struct tomsg_client *client) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	const int64_t tag = client->next_tag++;
	SEND_FMT0(client, tag, "logout");
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_LOGOUT,
	};
	return add_inflight(client, tag, event);
}

enum tomsg_retval tomsg_list_rooms(struct tomsg_client *client) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	const int64_t tag = client->next_tag++;
	SEND_FMT0(client, tag, "list_rooms");
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_LIST_ROOMS,
		.list_rooms.count = 0,
		.list_rooms.rooms = NULL,
	};
	return add_inflight(client, tag, event);
}

enum tomsg_retval tomsg_list_members(struct tomsg_client *client, const char *room_name) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	if (hasspacelf(room_name)) return TOMSG_ERR_SPACE;
	const int64_t tag = client->next_tag++;
	SEND_FMT(client, tag, "list_members %s", room_name);
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_LIST_MEMBERS,
		.list_members.room_name = strdup(room_name),
		.list_members.count = 0,
		.list_members.members = NULL,
	};
	return add_inflight(client, tag, event);
}

enum tomsg_retval tomsg_create_room(struct tomsg_client *client) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	const int64_t tag = client->next_tag++;
	SEND_FMT0(client, tag, "create_room");
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_CREATE_ROOM,
		.create_room.room_name = NULL,
	};
	return add_inflight(client, tag, event);
}

enum tomsg_retval tomsg_leave_room(
		struct tomsg_client *client, const char *room_name) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	if (hasspacelf(room_name)) return TOMSG_ERR_SPACE;
	const int64_t tag = client->next_tag++;
	SEND_FMT(client, tag, "leave_room %s", room_name);
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_LEAVE_ROOM,
		.leave_room.room_name = strdup(room_name),
	};
	return add_inflight(client, tag, event);
}

enum tomsg_retval tomsg_invite(
		struct tomsg_client *client, const char *room_name, const char *username) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	if (hasspacelf(room_name)) return TOMSG_ERR_SPACE;
	if (hasspacelf(username)) return TOMSG_ERR_SPACE;
	const int64_t tag = client->next_tag++;
	SEND_FMT(client, tag, "invite %s %s", room_name, username);
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_INVITE,
		.join.room_name = strdup(room_name),
		.join.username = strdup(username),
	};
	return add_inflight(client, tag, event);
}

enum tomsg_retval tomsg_send(
	struct tomsg_client *client, const char *room_name, const char *message,
	int64_t replyid,
	int64_t *tagp //output
) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	if (hasspacelf(room_name)) return TOMSG_ERR_SPACE;
	if (haslf(message)) return TOMSG_ERR_SPACE;
	const int64_t tag = client->next_tag++;
	if (replyid < 0) replyid = -1;
	SEND_FMT(client, tag, "send %s %" PRIi64 " %s", room_name, replyid, message);
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_SEND,
		.send.tag = tag,
	};
	if (tagp) *tagp = tag;
	return add_inflight(client, tag, event);
}

enum tomsg_retval tomsg_history(
		struct tomsg_client *client, const char *room_name,
		int64_t count, int64_t before_msgid) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	if (hasspacelf(room_name)) return TOMSG_ERR_SPACE;
	if (count < 0) count = 0;
	const int64_t tag = client->next_tag++;
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_HISTORY,
		.history.room_name = strdup(room_name),
		.history.count = 0,  // filled in when the first response is received
		.history.messages = NULL,
	};
	if (before_msgid < 0) {
		SEND_FMT(client, tag, "history %s %" PRIi64, room_name, count);
	} else {
		SEND_FMT(client, tag, "history_before %s %" PRIi64 " %" PRIi64,
				room_name, count, before_msgid);
	}
	return add_inflight(client, tag, event);
}

enum tomsg_retval tomsg_get_message(struct tomsg_client *client, int64_t msgid) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	if (msgid < -1) msgid = -1;
	const int64_t tag = client->next_tag++;
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_GET_MESSAGE,
	};
	SEND_FMT(client, tag, "get_message %" PRIi64, msgid);
	return add_inflight(client, tag, event);
}

enum tomsg_retval tomsg_ping(struct tomsg_client *client) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	const int64_t tag = client->next_tag++;
	SEND_FMT0(client, tag, "ping");
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_PING,
		.send.tag = tag,
	};
	return add_inflight(client, tag, event);
}

enum tomsg_retval tomsg_is_online(struct tomsg_client *client, const char *username) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	if (hasspacelf(username)) return TOMSG_ERR_SPACE;
	const int64_t tag = client->next_tag++;
	SEND_FMT(client, tag, "is_online %s", username);
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_IS_ONLINE,
		.is_online.username = strdup(username),
	};
	return add_inflight(client, tag, event);
}

enum tomsg_retval tomsg_user_active(struct tomsg_client *client, bool active) {
	if (!client->conn) return TOMSG_ERR_CLOSED;
	const int64_t tag = client->next_tag++;
	SEND_FMT(client, tag, "user_active %d", active);
	const struct tomsg_event event = (struct tomsg_event){
		.type = TOMSG_EV_USER_ACTIVE,
		.user_active.active = active,
	};
	return add_inflight(client, tag, event);
}